Spaces:
Paused
Paused
File size: 2,666 Bytes
23940b7 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
import os
from dotenv import load_dotenv
from supabase import create_client, Client
from typing import Optional
load_dotenv()
class SupabaseConnection:
"""Gerenciador de conexão com Supabase"""
_instance: Optional['SupabaseConnection'] = None
def __new__(cls):
"""Implementa Singleton para evitar múltiplas conexões"""
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self):
"""Inicializa a conexão com Supabase"""
if not hasattr(self, 'client'):
self.supabase_url = os.getenv("SUPABASE_URL")
self.supabase_key = os.getenv("SUPABASE_KEY")
self.service_role_key = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
if not all([self.supabase_url, self.supabase_key, self.service_role_key]):
raise ValueError(
"SUPABASE_URL, SUPABASE_KEY e SUPABASE_SERVICE_ROLE_KEY devem estar definidos no arquivo .env"
)
# Cliente normal para operações regulares
self.client = create_client(
supabase_url=self.supabase_url,
supabase_key=self.supabase_key
)
# Cliente admin para operações administrativas
self.admin_client = create_client(
supabase_url=self.supabase_url,
supabase_key=self.service_role_key
)
def get_client(self) -> Client:
"""Retorna o cliente Supabase regular"""
return self.client
def get_admin_client(self) -> Client:
"""Retorna o cliente Supabase com privilégios administrativos"""
return self.admin_client
def test_connection(self) -> bool:
"""Testa a conexão com Supabase"""
try:
# Tenta fazer uma query simples
print("🔍 Testando conexão com Supabase...")
response = self.client.table('users').select("*").limit(1).execute()
print("✅ Conexão Supabase bem sucedida!")
return True
except Exception as e:
print(f"❌ Erro na conexão Supabase: {str(e)}")
return False
def get_table(self, table_name: str):
"""
Retorna uma referência para uma tabela específica
Args:
table_name: Nome da tabela
Returns:
PostgrestFilterBuilder: Referência para a tabela
"""
return self.client.table(table_name) |