From 780bfe1c3074b1be68e44125b66afdcf3c37ae02 Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Mon, 23 Feb 2026 16:03:29 -0500 Subject: [PATCH] Refactor cache to store data rather than orm objects (#395) * fix: refactor cache to store data rather than orm objects * fix: Add Cache Version Keys --- src/crud/collection.py | 42 +++++++++++++++++++++++------- src/crud/peer.py | 31 ++++++++++++++++------ src/crud/session.py | 59 ++++++++++++++++++++++++++++++++---------- src/crud/workspace.py | 52 +++++++++++++++++++++++++++---------- 4 files changed, 138 insertions(+), 46 deletions(-) diff --git a/src/crud/collection.py b/src/crud/collection.py index fadbd145..f0496d50 100644 --- a/src/crud/collection.py +++ b/src/crud/collection.py @@ -5,6 +5,7 @@ from cashews import NOT_NONE from sqlalchemy import select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import make_transient_to_detached from src import models from src.cache.client import ( @@ -19,9 +20,9 @@ from src.exceptions import ConflictException, ResourceNotFoundException logger = getLogger(__name__) COLLECTION_CACHE_KEY_TEMPLATE = ( - "workspace:{workspace_name}:collection:{observer}:{observed}" + "v2:workspace:{workspace_name}:collection:{observer}:{observed}" ) -COLLECTION_LOCK_PREFIX = f"{get_cache_namespace()}:lock" +COLLECTION_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2" def collection_cache_key(workspace_name: str, observer: str, observed: str) -> str: @@ -53,14 +54,25 @@ async def _fetch_collection( workspace_name: str, observer: str, observed: str, -) -> models.Collection | None: - """Fetch a collection from the database.""" - return await db.scalar( +) -> dict[str, Any] | None: + """Fetch a collection from the database and return as a plain dict for safe caching.""" + obj = await db.scalar( select(models.Collection) .where(models.Collection.workspace_name == workspace_name) .where(models.Collection.observer == observer) .where(models.Collection.observed == observed) ) + if obj is None: + return None + return { + "id": obj.id, + "observer": obj.observer, + "observed": obj.observed, + "workspace_name": obj.workspace_name, + "h_metadata": obj.h_metadata, + "internal_metadata": obj.internal_metadata, + "created_at": obj.created_at, + } async def get_collection( @@ -85,11 +97,13 @@ async def get_collection( Raises: ResourceNotFoundException: If the collection does not exist """ - collection = await _fetch_collection(db, workspace_name, observer, observed) - if collection is None: + data = await _fetch_collection(db, workspace_name, observer, observed) + if data is None: raise ResourceNotFoundException("Collection not found") - # Merge cached object into session (cached objects are detached) - collection = await db.merge(collection, load=False) + # Reconstruct ORM object from cached dict and merge into session + obj = models.Collection(**data) + make_transient_to_detached(obj) + collection = await db.merge(obj, load=False) return collection @@ -118,7 +132,15 @@ async def get_or_create_collection( key = collection_cache_key(workspace_name, observer, observed) await safe_cache_set( key, - honcho_collection, + { + "id": honcho_collection.id, + "observer": honcho_collection.observer, + "observed": honcho_collection.observed, + "workspace_name": honcho_collection.workspace_name, + "h_metadata": honcho_collection.h_metadata, + "internal_metadata": honcho_collection.internal_metadata, + "created_at": honcho_collection.created_at, + }, expire=settings.CACHE.DEFAULT_TTL_SECONDS, ) diff --git a/src/crud/peer.py b/src/crud/peer.py index 8276a93e..7af73e9e 100644 --- a/src/crud/peer.py +++ b/src/crud/peer.py @@ -5,6 +5,7 @@ from cashews import NOT_NONE from sqlalchemy import Select, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import make_transient_to_detached from src import models, schemas from src.cache.client import cache, get_cache_namespace, safe_cache_delete @@ -17,8 +18,8 @@ from src.utils.types import GetOrCreateResult logger = getLogger(__name__) -PEER_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}:peer:{peer_name}" -PEER_LOCK_PREFIX = f"{get_cache_namespace()}:lock" +PEER_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}:peer:{peer_name}" +PEER_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2" def peer_cache_key(workspace_name: str, peer_name: str) -> str: @@ -152,12 +153,24 @@ async def _fetch_peer( db: AsyncSession, workspace_name: str, peer_name: str, -) -> models.Peer | None: - return await db.scalar( +) -> dict[str, Any] | None: + """Fetch a peer from the database and return as a plain dict for safe caching.""" + obj = await db.scalar( select(models.Peer) .where(models.Peer.workspace_name == workspace_name) .where(models.Peer.name == peer_name) ) + if obj is None: + return None + return { + "id": obj.id, + "name": obj.name, + "workspace_name": obj.workspace_name, + "h_metadata": obj.h_metadata, + "internal_metadata": obj.internal_metadata, + "configuration": obj.configuration, + "created_at": obj.created_at, + } async def get_peer( @@ -179,14 +192,16 @@ async def get_peer( Raises: ResourceNotFoundException: If the peer does not exist """ - existing_peer = await _fetch_peer(db, workspace_name, peer.name) - if existing_peer is None: + data = await _fetch_peer(db, workspace_name, peer.name) + if data is None: raise ResourceNotFoundException( f"Peer {peer.name} not found in workspace {workspace_name}" ) - # Merge cached object into session (cached objects are detached) - existing_peer = await db.merge(existing_peer, load=False) + # Reconstruct ORM object from cached dict and merge into session + obj = models.Peer(**data) + make_transient_to_detached(obj) + existing_peer = await db.merge(obj, load=False) return existing_peer diff --git a/src/crud/session.py b/src/crud/session.py index 668528f5..bac6ec23 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -10,6 +10,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.engine import CursorResult from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import make_transient_to_detached from sqlalchemy.types import BigInteger, Boolean from src import models, schemas @@ -43,8 +44,8 @@ class SessionDeletionResult: conclusions_deleted: int -SESSION_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}:session:{session_name}" -SESSION_LOCK_PREFIX = f"{get_cache_namespace()}:lock" +SESSION_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}:session:{session_name}" +SESSION_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2" def session_cache_key(workspace_name: str, session_name: str) -> str: @@ -74,12 +75,25 @@ async def _fetch_session( db: AsyncSession, workspace_name: str, session_name: str, -) -> models.Session | None: - return await db.scalar( +) -> dict[str, Any] | None: + """Fetch a session from the database and return as a plain dict for safe caching.""" + obj = await db.scalar( select(models.Session) .where(models.Session.workspace_name == workspace_name) .where(models.Session.name == session_name) ) + if obj is None: + return None + return { + "id": obj.id, + "name": obj.name, + "workspace_name": obj.workspace_name, + "is_active": obj.is_active, + "h_metadata": obj.h_metadata, + "internal_metadata": obj.internal_metadata, + "configuration": obj.configuration, + "created_at": obj.created_at, + } def count_observers_in_config( @@ -143,11 +157,14 @@ async def get_or_create_session( if not session.name: raise ValueError("Session name must be provided") - honcho_session = await _fetch_session(db, workspace_name, session.name) + session_data = await _fetch_session(db, workspace_name, session.name) - # Merge cached object into session if it exists (cached objects are detached) - if honcho_session is not None: - honcho_session = await db.merge(honcho_session, load=False) + # Reconstruct and merge cached dict into session if it exists + honcho_session: models.Session | None = None + if session_data is not None: + obj = models.Session(**session_data) + make_transient_to_detached(obj) + honcho_session = await db.merge(obj, load=False) # Reject operations on inactive sessions (marked for deletion) if not honcho_session.is_active: @@ -239,7 +256,18 @@ async def get_or_create_session( if needs_cache_update: cache_key = session_cache_key(workspace_name, session.name) await safe_cache_set( - cache_key, honcho_session, expire=settings.CACHE.DEFAULT_TTL_SECONDS + cache_key, + { + "id": honcho_session.id, + "name": honcho_session.name, + "workspace_name": honcho_session.workspace_name, + "is_active": honcho_session.is_active, + "h_metadata": honcho_session.h_metadata, + "internal_metadata": honcho_session.internal_metadata, + "configuration": honcho_session.configuration, + "created_at": honcho_session.created_at, + }, + expire=settings.CACHE.DEFAULT_TTL_SECONDS, ) logger.debug( "Session %s cache updated in workspace %s", session.name, workspace_name @@ -271,21 +299,24 @@ async def get_session( Raises: ResourceNotFoundException: If the session does not exist or is inactive """ - session = await _fetch_session(db, workspace_name, session_name) + data = await _fetch_session(db, workspace_name, session_name) - if session is None: + if data is None: raise ResourceNotFoundException( f"Session {session_name} not found in workspace {workspace_name}" ) # Check if session is active (unless include_inactive is True) - if not include_inactive and not session.is_active: + # Check on the dict before constructing the ORM object + if not include_inactive and not data["is_active"]: raise ResourceNotFoundException( f"Session {session_name} not found in workspace {workspace_name}" ) - # Merge cached object into session (cached objects are detached) - session = await db.merge(session, load=False) + # Reconstruct ORM object from cached dict and merge into session + obj = models.Session(**data) + make_transient_to_detached(obj) + session = await db.merge(obj, load=False) return session diff --git a/src/crud/workspace.py b/src/crud/workspace.py index f759543e..b77223c0 100644 --- a/src/crud/workspace.py +++ b/src/crud/workspace.py @@ -6,6 +6,7 @@ from cashews import NOT_NONE from sqlalchemy import Select, delete, exists, func, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import make_transient_to_detached from src import models, schemas from src.cache.client import ( @@ -34,8 +35,8 @@ class WorkspaceDeletionResult: conclusions_deleted: int -WORKSPACE_CACHE_KEY_TEMPLATE = "workspace:{workspace_name}" -WORKSPACE_LOCK_PREFIX = f"{get_cache_namespace()}:lock" +WORKSPACE_CACHE_KEY_TEMPLATE = "v2:workspace:{workspace_name}" +WORKSPACE_LOCK_PREFIX = f"{get_cache_namespace()}:lock:v2" def workspace_cache_key(workspace_name: str) -> str: @@ -60,11 +61,21 @@ def workspace_cache_key(workspace_name: str) -> str: ) async def _fetch_workspace( db: AsyncSession, workspace_name: str -) -> models.Workspace | None: - """Fetch a workspace from the database.""" - return await db.scalar( +) -> dict[str, Any] | None: + """Fetch a workspace from the database and return as a plain dict for safe caching.""" + obj = await db.scalar( select(models.Workspace).where(models.Workspace.name == workspace_name) ) + if obj is None: + return None + return { + "id": obj.id, + "name": obj.name, + "h_metadata": obj.h_metadata, + "internal_metadata": obj.internal_metadata, + "configuration": obj.configuration, + "created_at": obj.created_at, + } async def get_or_create_workspace( @@ -91,12 +102,14 @@ async def get_or_create_workspace( raise ValueError("Workspace name must be provided") # Check if workspace already exists - existing_workspace = await _fetch_workspace(db, workspace.name) - if existing_workspace is not None: + data = await _fetch_workspace(db, workspace.name) + if data is not None: # Workspace already exists logger.debug("Found existing workspace: %s", workspace.name) - # Merge cached object into session (cached objects are detached) - existing_workspace = await db.merge(existing_workspace, load=False) + # Reconstruct ORM object from cached dict and merge into session + obj = models.Workspace(**data) + make_transient_to_detached(obj) + existing_workspace = await db.merge(obj, load=False) return GetOrCreateResult(existing_workspace, created=False) # Workspace doesn't exist, create a new one @@ -114,7 +127,16 @@ async def get_or_create_workspace( cache_key = workspace_cache_key(workspace.name) await safe_cache_set( - cache_key, honcho_workspace, expire=settings.CACHE.DEFAULT_TTL_SECONDS + cache_key, + { + "id": honcho_workspace.id, + "name": honcho_workspace.name, + "h_metadata": honcho_workspace.h_metadata, + "internal_metadata": honcho_workspace.internal_metadata, + "configuration": honcho_workspace.configuration, + "created_at": honcho_workspace.created_at, + }, + expire=settings.CACHE.DEFAULT_TTL_SECONDS, ) return GetOrCreateResult(honcho_workspace, created=True) except IntegrityError: @@ -159,13 +181,15 @@ async def get_workspace( Raises: ResourceNotFoundException: If the workspace does not exist """ - existing_workspace = await _fetch_workspace(db, workspace_name) + data = await _fetch_workspace(db, workspace_name) - if existing_workspace is None: + if data is None: raise ResourceNotFoundException(f"Workspace {workspace_name} not found") - # Merge cached object into session (cached objects are detached) - existing_workspace = await db.merge(existing_workspace, load=False) + # Reconstruct ORM object from cached dict and merge into session + obj = models.Workspace(**data) + make_transient_to_detached(obj) + existing_workspace = await db.merge(obj, load=False) return existing_workspace