File size: 2,660 Bytes
544c0b2 | 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 | from supabase import create_client, Client
import os
from typing import Optional, Any, Dict, List
class SupabaseClient:
_instance = None
def __new__(cls):
if cls._instance is None:
cls._instance = super(SupabaseClient, cls).__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
try:
self.client: Client = create_client(
os.environ.get('SUP_URL', ''),
os.environ.get('SUP_KEY', '')
)
self._initialized = True
except Exception as e:
raise Exception(f"Supabase initialization failed: {e}")
def select(self, table: str, columns: str = "*") -> List[Dict]:
"""Select data from specified table"""
try:
response = self.client.table(table).select(columns).execute()
return response.data
except Exception as e:
raise Exception(f"Supabase select error: {e}")
def insert(self, table: str, data: Dict) -> Dict:
"""Insert data into specified table"""
try:
response = self.client.table(table).insert(data).execute()
return response.data
except Exception as e:
raise Exception(f"Supabase insert error: {e}")
def update(self, table: str, data: Dict, query: Dict) -> Dict:
"""Update data in specified table based on query"""
try:
response = self.client.table(table).update(data).match(query).execute()
return response.data
except Exception as e:
raise Exception(f"Supabase update error: {e}")
def delete(self, table: str, query: Dict) -> bool:
"""Delete data from specified table based on query"""
try:
self.client.table(table).delete().match(query).execute()
return True
except Exception as e:
raise Exception(f"Supabase delete error: {e}")
def upsert(self, table: str, data: Dict) -> Dict:
"""Insert or update data in specified table"""
try:
response = self.client.table(table).upsert(data).execute()
return response.data
except Exception as e:
raise Exception(f"Supabase upsert error: {e}")
def rpc(self, function: str, params: Optional[Dict] = None) -> Any:
"""Call a postgres function"""
try:
response = self.client.rpc(function, params).execute()
return response.data
except Exception as e:
raise Exception(f"Supabase rpc error: {e}") |