RomanNum3ral
/
honcho
mirror of https://github.com/plastic-labs/honcho.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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
This commit is contained in:
Vineeth Voruganti 2026-02-23 16:03:29 -05:00 committed by GitHub
parent 78df86dc66
commit 780bfe1c30
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 138 additions and 46 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
src/crud/collection.py
View File

@ -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,
)

31
src/crud/peer.py
View File

@ -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

59
src/crud/session.py
View File

@ -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

52
src/crud/workspace.py
View File

@ -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