bugfix: use new DB_SCHEMA config label in migrations; use internal_metadata on documents in CRUD (#143)

* fix: swap to DB_SCHEMA; swap to internal_metadata

* fix: use settings for last remaining env vars

.env.template

@@ -10,6 +10,7 @@
 LOG_LEVEL=INFO
 FASTAPI_HOST=0.0.0.0
 FASTAPI_PORT=8000
+# SESSION_PEERS_LIMIT=10
 
 # =============================================================================
 # Database Settings (REQUIRED)
@@ -107,5 +108,7 @@ LLM_ANTHROPIC_API_KEY=your-anthropic-api-key-here
 # Sentry error tracking
 # SENTRY_ENABLED=false
 # SENTRY_DSN=your-sentry-dsn-here
+# SENTRY_RELEASE=your-release-semver
+# SENTRY_ENVIRONMENT=string-that-labels-deployment (default "development")
 # SENTRY_TRACES_SAMPLE_RATE=0.1
 # SENTRY_PROFILES_SAMPLE_RATE=0.1

migrations/versions/66e63cf2cf77_add_indexes_to_documents_table.py

@@ -7,17 +7,18 @@ Create Date: 2025-05-19 17:00:18.151735
 """
 
 from collections.abc import Sequence
-from os import getenv
 
 from alembic import op
 from sqlalchemy import text
 
+from src.config import settings
+
 # revision identifiers, used by Alembic.
 revision: str = "66e63cf2cf77"
 down_revision: str | None = "20f89a421aff"
 branch_labels: str | Sequence[str] | None = None
 depends_on: str | Sequence[str] | None = None
-schema = getenv("DATABASE_SCHEMA", "public")
+schema = settings.DB.SCHEMA
 
 
 def upgrade() -> None:

migrations/versions/d429de0e5338_adopt_peer_paradigm.py

@@ -8,7 +8,6 @@ Create Date: 2025-06-09 15:16:38.164067
 
 from collections.abc import Sequence
 from contextlib import suppress
-from os import getenv
 
 import sqlalchemy as sa
 import tiktoken
@@ -24,6 +23,7 @@ from migrations.utils import (
     index_exists,
     table_exists,
 )
+from src.config import settings
 
 # revision identifiers, used by Alembic.
 revision: str = "d429de0e5338"
@@ -34,7 +34,7 @@ depends_on: str | Sequence[str] | None = None
 
 def upgrade() -> None:
     """Upgrade database schema to adopt peer paradigm."""
-    schema = getenv("DATABASE_SCHEMA", "public")
+    schema = settings.DB.SCHEMA
     inspector = sa.inspect(op.get_bind())
 
     # Step 1: Rename tables
@@ -77,7 +77,7 @@ def upgrade() -> None:
 
 def downgrade() -> None:
     """Downgrade database schema to reverse peer paradigm adoption."""
-    schema = getenv("DATABASE_SCHEMA", "public")
+    schema = settings.DB.SCHEMA
     inspector = sa.inspect(op.get_bind())
 
     # Step 1: Add back app_id, user_id to peers and sessions

src/agent.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import os
 
 from langfuse.decorators import langfuse_context, observe  # pyright: ignore
 from mirascope import llm
@@ -216,8 +215,8 @@ async def chat(
     langfuse_context.update_current_trace(
         session_id=session_name,
         user_id=peer_name,
-        release=os.getenv("SENTRY_RELEASE"),
-        metadata={"environment": os.getenv("SENTRY_ENVIRONMENT")},
+        release=settings.SENTRY.RELEASE,
+        metadata={"environment": settings.SENTRY.ENVIRONMENT},
     )
 
     # Use streaming or non-streaming response based on the request

src/config.py

@@ -156,6 +156,8 @@ class SentrySettings(HonchoSettings):
 
     ENABLED: bool = False
     DSN: str | None = None
+    RELEASE: str | None = None  # TODO maybe centralize this with release number
+    ENVIRONMENT: str = "development"
     TRACES_SAMPLE_RATE: Annotated[float, Field(default=0.1, ge=0.0, le=1.0)] = 0.1
     PROFILES_SAMPLE_RATE: Annotated[float, Field(default=0.1, ge=0.0, le=1.0)] = 0.1
 
@@ -243,6 +245,7 @@ class AppSettings(HonchoSettings):
     LOG_LEVEL: str = "INFO"
     FASTAPI_HOST: str = "0.0.0.0"
     FASTAPI_PORT: Annotated[int, Field(default=8000, gt=0, le=65535)] = 8000
+    SESSION_PEERS_LIMIT: Annotated[int, Field(default=10, gt=0)] = 10
 
     # Nested settings models
     DB: DBSettings = Field(default_factory=DBSettings)

src/crud.py

@@ -1,4 +1,3 @@
-import os
 from collections.abc import Sequence
 from logging import getLogger
 from typing import Any, final
@@ -42,8 +41,6 @@ logger = getLogger(__name__)
 
 USER_REPRESENTATION_METADATA_KEY = "user_representation"
 
-SESSION_PEERS_LIMIT = int(os.getenv("SESSION_PEERS_LIMIT", 10))
-
 ########################################################
 # workspace methods
 ########################################################
@@ -391,9 +388,12 @@ async def get_or_create_session(
 
     # Check if session already exists
     if honcho_session is None:
-        if session.peer_names and len(session.peer_names) > SESSION_PEERS_LIMIT:
+        if (
+            session.peer_names
+            and len(session.peer_names) > settings.SESSION_PEERS_LIMIT
+        ):
             raise ValueError(
-                f"Cannot create session {session.name} with {len(session.peer_names)} peers. Maximum allowed is {SESSION_PEERS_LIMIT} peers per session."
+                f"Cannot create session {session.name} with {len(session.peer_names)} peers. Maximum allowed is {settings.SESSION_PEERS_LIMIT} peers per session."
             )
 
         # Create honcho session
@@ -780,9 +780,9 @@ async def set_peers_for_session(
         ResourceNotFoundException: If the session does not exist
     """
     # Validate peer limit before making any changes
-    if len(peer_names) > SESSION_PEERS_LIMIT:
+    if len(peer_names) > settings.SESSION_PEERS_LIMIT:
         raise ValueError(
-            f"Cannot set {len(peer_names)} peers for session {session_name}. Maximum allowed is {SESSION_PEERS_LIMIT} peers per session."
+            f"Cannot set {len(peer_names)} peers for session {session_name}. Maximum allowed is {settings.SESSION_PEERS_LIMIT} peers per session."
         )
 
     # Verify session exists
@@ -871,9 +871,9 @@ async def _get_or_add_peers_to_session(
     existing_peer_names = result.scalars().all()
 
     new_peers = [name for name in peer_names if name not in existing_peer_names]
-    if len(new_peers) + len(existing_peer_names) > SESSION_PEERS_LIMIT:
+    if len(new_peers) + len(existing_peer_names) > settings.SESSION_PEERS_LIMIT:
         raise ValueError(
-            f"Cannot add {len(new_peers)} peer(s). Session already has {len(existing_peer_names)} peer(s) with {SESSION_PEERS_LIMIT} peers per session."
+            f"Cannot add {len(new_peers)} peer(s). Session already has {len(existing_peer_names)} peer(s) with {settings.SESSION_PEERS_LIMIT} peers per session."
         )
 
     # Use upsert to handle both new peers and rejoining peers
@@ -1549,7 +1549,7 @@ async def query_documents(
             models.Document.embedding.cosine_distance(embedding_query) < max_distance
         )
     if filter is not None:
-        stmt = stmt.where(models.Document.h_metadata.contains(filter))
+        stmt = stmt.where(models.Document.internal_metadata.contains(filter))
     stmt = stmt.limit(top_k).order_by(
         models.Document.embedding.cosine_distance(embedding_query)
     )

src/routers/messages.py

@@ -1,5 +1,4 @@
 import logging
-import os
 from typing import Any
 
 from fastapi import APIRouter, BackgroundTasks, Body, Depends, Path, Query
@@ -9,6 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.sql import insert
 
 from src import crud, schemas
+from src.config import settings
 from src.dependencies import db, tracked_db
 from src.exceptions import ResourceNotFoundException
 from src.models import QueueItem
@@ -255,7 +255,7 @@ async def enqueue(payload: list[dict[str, Any]]):
 
         except Exception as e:
             logger.error(f"Failed to enqueue messages: {str(e)}", exc_info=True)
-            if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
+            if settings.SENTRY.ENABLED:
                 import sentry_sdk
 
                 sentry_sdk.capture_exception(e)