From 9378235337dc4229e847a9169e782c426bae413e Mon Sep 17 00:00:00 2001 From: Vineeth Voruganti <13438633+VVoruganti@users.noreply.github.com> Date: Mon, 23 Feb 2026 11:36:25 -0500 Subject: [PATCH] fix: refactor cache to store data rather than orm objects --- src/crud/collection.py | 38 +++++++++++++++++++++++------ src/crud/peer.py | 27 ++++++++++++++++----- src/crud/session.py | 55 +++++++++++++++++++++++++++++++++--------- src/crud/workspace.py | 48 +++++++++++++++++++++++++++--------- 4 files changed, 130 insertions(+), 38 deletions(-) diff --git a/src/crud/collection.py b/src/crud/collection.py index fadbd145..e5ced71d 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 ( @@ -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..22c33603 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 @@ -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 9657413e..2d0f857c 100644 --- a/src/crud/session.py +++ b/src/crud/session.py @@ -8,6 +8,7 @@ from sqlalchemy import Select, and_, case, cast, delete, func, insert, select, u from sqlalchemy.dialects.postgresql import insert as pg_insert 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 @@ -72,12 +73,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( @@ -141,11 +155,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: @@ -237,7 +254,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 @@ -269,21 +297,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..c6a10500 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 ( @@ -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