Spaces:
Running
Running
File size: 2,888 Bytes
24480a0 | 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 | """
backend/memory/supabase_backend.py — Supabase implementation of MemoryBackend.
"""
from typing import Any, Dict, List, Optional
from .memory_backend import MemoryBackend
class SupabaseMemoryBackend(MemoryBackend):
"""
Wrappa il client Supabase esistente in api.state per l'uso nel MemorySynchronizer.
Il client viene risolto late-binding al momento di ogni operazione (non all'import)
per supportare configurazioni in cui Supabase si connette dopo l'import del modulo.
"""
def __init__(self):
self.is_connected = self._client() is not None
def _client(self):
"""Risolve il client Supabase corrente — None se non configurato."""
try:
from api.state import _sb
return _sb
except Exception:
return None
def _require_client(self):
"""Lancia RuntimeError se il client non è disponibile."""
c = self._client()
if c is None:
raise RuntimeError(
"Supabase client non disponibile: imposta SUPABASE_URL e SUPABASE_KEY "
"nelle variabili HF Space/env prima di usare SupabaseMemoryBackend."
)
return c
async def connect(self, config: Dict[str, Any]) -> None:
"""Il client è già connesso via api.state."""
self.is_connected = self._client() is not None
async def disconnect(self) -> None:
"""Nessuna operazione di chiusura necessaria per il client HTTP stateless."""
self.is_connected = False
async def add_memory(self, user_id: str, memory_data: Dict[str, Any], memory_id: Optional[str] = None) -> str:
c = self._require_client()
data = {**memory_data, "user_id": user_id}
if memory_id:
data["id"] = memory_id
result = c.table("memories").insert(data).execute()
return result.data[0]["id"]
async def get_memory(self, user_id: str, memory_id: str) -> Optional[Dict[str, Any]]:
c = self._require_client()
result = c.table("memories").select("*").eq("user_id", user_id).eq("id", memory_id).execute()
return result.data[0] if result.data else None
async def list_memories(self, user_id: str, limit: int = 100, offset: int = 0) -> List[Dict[str, Any]]:
c = self._require_client()
result = c.table("memories").select("*").eq("user_id", user_id).range(offset, offset + limit).execute()
return result.data
async def update_memory(self, user_id: str, memory_id: str, new_data: Dict[str, Any]) -> bool:
c = self._require_client()
result = c.table("memories").update(new_data).eq("user_id", user_id).eq("id", memory_id).execute()
return len(result.data) > 0
async def delete_memory(self, user_id: str, memory_id: str) -> bool:
c = self._require_client()
result = c.table("memories").delete().eq("user_id", user_id).eq("id", memory_id).execute()
return len(result.data) > 0
|