Spaces:
Running
Running
File size: 2,406 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 72 73 74 75 76 77 78 | """
backend/memory/memory_backend.py — Abstract interface for memory backends.
Defines the MemoryBackend contract implemented by both the Supabase adapter
and the OracleMemoryAdapter. All concrete backends must implement every method.
"""
from __future__ import annotations
import abc
from typing import Any, Dict, List, Optional
class MemoryBackend(abc.ABC):
"""Abstract base class for all memory storage backends."""
# is_connected lets callers guard operations without catching ConnectionError.
is_connected: bool = False
@abc.abstractmethod
async def connect(self, config: Dict[str, Any]) -> None:
"""Establish a connection to the backend."""
@abc.abstractmethod
async def disconnect(self) -> None:
"""Close the connection to the backend."""
@abc.abstractmethod
async def add_memory(
self,
user_id: str,
memory_data: Dict[str, Any],
memory_id: Optional[str] = None,
) -> str:
"""
Insert a new memory entry for *user_id*.
Args:
user_id: Owning user's identifier.
memory_data: Arbitrary JSON-serialisable payload.
memory_id: Optional caller-supplied ID (must be honoured when provided
so that primary and secondary backends stay in sync).
Returns:
The memory_id that was stored (either caller-supplied or generated).
"""
@abc.abstractmethod
async def get_memory(
self, user_id: str, memory_id: str
) -> Optional[Dict[str, Any]]:
"""Return the memory payload or None if not found."""
@abc.abstractmethod
async def update_memory(
self, user_id: str, memory_id: str, new_data: Dict[str, Any]
) -> bool:
"""
Replace the payload of an existing memory entry.
Returns:
True if a row was updated, False if the entry was not found.
"""
@abc.abstractmethod
async def delete_memory(self, user_id: str, memory_id: str) -> bool:
"""
Remove a memory entry.
Returns:
True if a row was deleted, False if the entry was not found.
"""
@abc.abstractmethod
async def list_memories(
self, user_id: str, limit: int = 100, offset: int = 0
) -> List[Dict[str, Any]]:
"""Return a paginated list of memory payloads for *user_id*."""
|