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

feat (config): Using pydantic settings for managing settings values across the project

This commit is contained in:
Vineeth Voruganti 2025-05-27 15:52:25 -04:00
parent 91b7436fb9
commit a078783800
11 changed files with 1469 additions and 1263 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
.gitignore vendored
View File

@ -176,4 +176,6 @@ supabase/
docs/node_modules
timing_logs.csv
timing_logs.csv
config.toml

76
config.toml.example Normal file
View File

@ -0,0 +1,76 @@
# Honcho Configuration File
# This file demonstrates all available configuration options.
# Copy this to config.toml and modify as needed.
# Environment variables will override these values.
# Application-level settings
[app]
LOG_LEVEL = "INFO"
FASTAPI_HOST = "0.0.0.0"
FASTAPI_PORT = 8000
# Database settings
[db]
CONNECTION_URI = "postgresql+psycopg://postgres:postgres@localhost:5432/postgres"
SCHEMA = "public"
POOL_PRE_PING = true
POOL_SIZE = 10
MAX_OVERFLOW = 20
POOL_TIMEOUT = 30 # seconds
POOL_RECYCLE = 300 # seconds
POOL_USE_LIFO = true
SQL_DEBUG = false
# Authentication settings
[auth]
USE_AUTH = true
JWT_SECRET = "your-secret-key-here" # Must be set if USE_AUTH is true
# Sentry settings
[sentry]
ENABLED = false
DSN = ""
TRACES_SAMPLE_RATE = 0.1
PROFILES_SAMPLE_RATE = 0.1
# OpenTelemetry settings
[opentelemetry]
ENABLED = false
# LLM settings
[llm]
DEFAULT_MAX_TOKENS = 1000
DEFAULT_TEMPERATURE = 0.0
# Dialectic specific
DIALECTIC_PROVIDER = "anthropic"
DIALECTIC_MODEL = "claude-3-7-sonnet-20250219"
# Query Generation specific
QUERY_GENERATION_PROVIDER = "groq"
QUERY_GENERATION_MODEL = "llama-3.1-8b-instant"
# Summarization specific
SUMMARY_PROVIDER = "gemini"
SUMMARY_MODEL = "gemini-2.0-flash-lite"
SUMMARY_MAX_TOKENS_SHORT = 1000
SUMMARY_MAX_TOKENS_LONG = 2000
# Agent settings
[agent]
SEMANTIC_SEARCH_TOP_K = 10
SEMANTIC_SEARCH_MAX_DISTANCE = 0.85
TOM_INFERENCE_METHOD = "single_prompt"
# Deriver settings
[deriver]
WORKERS = 1
STALE_SESSION_TIMEOUT_MINUTES = 5
POLLING_SLEEP_INTERVAL_SECONDS = 1.0
TOM_METHOD = "single_prompt"
USER_REPRESENTATION_METHOD = "long_term"
# History settings
[history]
MESSAGES_PER_SHORT_SUMMARY = 20
MESSAGES_PER_LONG_SUMMARY = 60

1
pyproject.toml
View File

@ -26,6 +26,7 @@ dependencies = [
"pyjwt>=2.10.0",
"google-genai>=1.10.0",
"pydantic-settings>=2.9.1",
"tomli>=2.0.0; python_version < '3.11'",
]
[tool.uv]
dev-dependencies = [

26
src/agent.py
View File

@ -24,7 +24,9 @@ from src.utils.model_client import ModelClient, ModelProvider
# Configure logging
logger = logging.getLogger(__name__)
USER_REPRESENTATION_METAMESSAGE_TYPE = "honcho_user_representation"
USER_REPRESENTATION_METAMESSAGE_TYPE = (
settings.AGENT.USER_REPRESENTATION_METAMESSAGE_TYPE
)
QUERY_GENERATION_SYSTEM = """Given this query about a user, generate 3 focused search queries that would help retrieve relevant facts about the user.
Each query should focus on a specific aspect related to the original query, rephrased to maximize semantic search effectiveness.
@ -43,7 +45,7 @@ class Dialectic:
self.chat_history = chat_history
self.client = ModelClient(
provider=ModelProvider(settings.LLM.DIALECTIC_PROVIDER),
model=settings.LLM.DIALECTIC_MODEL
model=settings.LLM.DIALECTIC_MODEL,
)
self.system_prompt = """You are operating as a context service that helps maintain psychological understanding of users across applications. Alongside a query, you'll receive: 1) previously collected psychological context about the user that I've maintained, 2) a series of long-term facts about the user, and 3) their current conversation/interaction from the requesting application. Your goal is to analyze this information and provide theory-of-mind insights that help applications personalize their responses. Please respond in a brief, matter-of-fact, and appropriate manner to convey as much relevant information to the application based on its query and the user's most recent message. You are encouraged to provide any context from the provided resources that helps provide a more complete or nuanced understanding of the user, as long as it is somewhat relevant to the query. If the context provided doesn't help address the query, write absolutely NOTHING but "None"."""
@ -74,7 +76,9 @@ class Dialectic:
logger.debug("Calling model for generation")
model_start = asyncio.get_event_loop().time()
response = await self.client.generate(
messages=[message], system=self.system_prompt, max_tokens=settings.LLM.DEFAULT_MAX_TOKENS
messages=[message],
system=self.system_prompt,
max_tokens=settings.LLM.DEFAULT_MAX_TOKENS,
)
model_time = asyncio.get_event_loop().time() - model_start
logger.debug(
@ -112,7 +116,9 @@ class Dialectic:
logger.debug("Calling model for streaming")
model_start = asyncio.get_event_loop().time()
stream = await self.client.stream(
messages=[message], system=self.system_prompt, max_tokens=settings.LLM.DEFAULT_MAX_TOKENS
messages=[message],
system=self.system_prompt,
max_tokens=settings.LLM.DEFAULT_MAX_TOKENS,
)
stream_setup_time = asyncio.get_event_loop().time() - model_start
@ -284,7 +290,10 @@ async def get_long_term_facts(
collection_id=collection_id,
)
facts = await query_embedding_store.get_relevant_facts(
search_query, top_k=10, max_distance=0.85
search_query,
top_k=settings.AGENT.SEMANTIC_SEARCH_TOP_K,
max_distance=settings.AGENT.SEMANTIC_SEARCH_MAX_DISTANCE,
)
query_time = asyncio.get_event_loop().time() - query_start
logger.debug(f"Query {i + 1} retrieved {len(facts)} facts in {query_time:.2f}s")
return facts
@ -324,7 +333,10 @@ async def run_tom_inference(chat_history: str, session_id: str) -> str:
# Get chat history length to determine if this is a new conversation
tom_inference_response = await get_tom_inference(
chat_history, session_id, method=settings.AGENT.TOM_INFERENCE_METHOD, user_representation=""
chat_history,
session_id,
method=settings.AGENT.TOM_INFERENCE_METHOD,
user_representation="",
)
# Extract the prediction from the response
@ -357,7 +369,7 @@ async def generate_semantic_queries(query: str) -> list[str]:
# Create a new model client
client = ModelClient(
provider=ModelProvider(settings.LLM.QUERY_GENERATION_PROVIDER),
model=settings.LLM.QUERY_GENERATION_MODEL
model=settings.LLM.QUERY_GENERATION_MODEL,
)
# Prepare the messages for Anthropic

214
src/config.py
View File

@ -1,46 +1,166 @@
import os
from typing import Optional
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from pydantic_settings import BaseSettings, SettingsConfigDict
import tomllib
from dotenv import load_dotenv
from pydantic_settings import (
BaseSettings,
PydanticBaseSettingsSource,
SettingsConfigDict,
)
# Load .env file for local development.
# Make sure this is called before AppSettings is instantiated if you rely on .env for AppSettings construction.
load_dotenv()
class DBSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='DB_')
CONNECTION_URI: str = "postgresql+psycopg://postgres:postgres@localhost:5432/postgres"
def load_toml_config(config_path: str = "config.toml") -> Dict[str, Any]:
"""Load configuration from TOML file if it exists."""
config_file = Path(config_path)
if config_file.exists():
try:
with open(config_file, "rb") as f:
return tomllib.load(f)
except Exception as e:
print(f"Warning: Failed to load {config_path}: {e}")
return {}
return {}
# Load TOML config once
TOML_CONFIG = load_toml_config()
class TomlConfigSettingsSource(PydanticBaseSettingsSource):
"""Custom settings source for loading from TOML file."""
def __init__(self, settings_cls: type[BaseSettings]):
super().__init__(settings_cls)
def get_field_value(
self, field_name: str, field_info: Any
) -> Tuple[Any, str, bool]:
# Get the env_prefix from the model config
prefix = self.settings_cls.model_config.get("env_prefix", "")
if prefix.endswith("_"):
prefix = prefix[:-1]
# Map prefixes to TOML sections
section_map = {
"DB": "db",
"AUTH": "auth",
"SENTRY": "sentry",
"OPENTELEMETRY": "opentelemetry",
"LLM": "llm",
"AGENT": "agent",
"DERIVER": "deriver",
"HISTORY": "history",
"": "app", # For AppSettings with no prefix
}
section = section_map.get(prefix, prefix.lower())
toml_data = TOML_CONFIG.get(section, {})
# Try different case variations
field_value = toml_data.get(field_name.lower())
if field_value is None:
field_value = toml_data.get(field_name.upper())
if field_value is None:
field_value = toml_data.get(field_name)
return field_value, field_name, False
def __call__(self) -> Dict[str, Any]:
# Get the env_prefix from the model config
prefix = self.settings_cls.model_config.get("env_prefix", "")
if prefix.endswith("_"):
prefix = prefix[:-1]
# Map prefixes to TOML sections
section_map = {
"DB": "db",
"AUTH": "auth",
"SENTRY": "sentry",
"OPENTELEMETRY": "opentelemetry",
"LLM": "llm",
"AGENT": "agent",
"DERIVER": "deriver",
"HISTORY": "history",
"": "app", # For AppSettings with no prefix
}
section = section_map.get(prefix, prefix.lower())
toml_data = TOML_CONFIG.get(section, {})
# Convert keys to uppercase to match field names
result = {}
for key, value in toml_data.items():
result[key.upper()] = value
return result
class TomlSettings(BaseSettings):
"""Base settings class that loads from TOML config first, then env vars."""
@classmethod
def settings_customise_sources(
cls,
settings_cls,
init_settings,
env_settings,
dotenv_settings,
file_secret_settings,
):
# Return sources in priority order (first is lowest priority)
return (
init_settings,
TomlConfigSettingsSource(settings_cls),
dotenv_settings,
env_settings,
file_secret_settings,
)
class DBSettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="DB_")
CONNECTION_URI: str = (
"postgresql+psycopg://postgres:postgres@localhost:5432/postgres"
)
SCHEMA: str = "public"
POOL_PRE_PING: bool = True
POOL_SIZE: int = 10
MAX_OVERFLOW: int = 20
POOL_TIMEOUT: int = 30 # seconds
POOL_RECYCLE: int = 300 # seconds
POOL_TIMEOUT: int = 30 # seconds
POOL_RECYCLE: int = 300 # seconds
POOL_USE_LIFO: bool = True
SQL_DEBUG: bool = False
class AuthSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='AUTH_')
class AuthSettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="AUTH_")
USE_AUTH: bool = True
JWT_SECRET: Optional[str] = None # Must be set if USE_AUTH is true
JWT_SECRET: Optional[str] = None # Must be set if USE_AUTH is true
class SentrySettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='SENTRY_')
class SentrySettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="SENTRY_")
ENABLED: bool = False
DSN: Optional[str] = None
TRACES_SAMPLE_RATE: float = 0.1
PROFILES_SAMPLE_RATE: float = 0.1
class OpenTelemetrySettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='OPENTELEMETRY_')
class OpenTelemetrySettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="OPENTELEMETRY_")
ENABLED: bool = False
class LLMSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='LLM_')
class LLMSettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="LLM_")
# General LLM settings
DEFAULT_MAX_TOKENS: int = 1000
DEFAULT_TEMPERATURE: float = 0.0
@ -57,21 +177,28 @@ class LLMSettings(BaseSettings):
# Summarization specific
SUMMARY_PROVIDER: str = "gemini"
SUMMARY_MODEL: str = "gemini-2.0-flash-lite" # Consider specific model version if needed
SUMMARY_MODEL: str = (
"gemini-2.0-flash-lite" # Consider specific model version if needed
)
SUMMARY_MAX_TOKENS_SHORT: int = 1000
SUMMARY_MAX_TOKENS_LONG: int = 2000
# SUMMARY_SYSTEM_PROMPT_SHORT_FILE: Optional[str] = "prompts/summary_short_system.txt"
# SUMMARY_SYSTEM_PROMPT_LONG_FILE: Optional[str] = "prompts/summary_long_system.txt"
class AgentSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='AGENT_')
class AgentSettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="AGENT_")
SEMANTIC_SEARCH_TOP_K: int = 10
SEMANTIC_SEARCH_MAX_DISTANCE: float = 0.85 # Max distance for semantic search relevance
SEMANTIC_SEARCH_MAX_DISTANCE: float = (
0.85 # Max distance for semantic search relevance
)
TOM_INFERENCE_METHOD: str = "single_prompt"
USER_REPRESENTATION_METAMESSAGE_TYPE: str = "honcho_user_representation"
class DeriverSettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='DERIVER_')
class DeriverSettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="DERIVER_")
WORKERS: int = 1
STALE_SESSION_TIMEOUT_MINUTES: int = 5
@ -79,13 +206,18 @@ class DeriverSettings(BaseSettings):
TOM_METHOD: str = "single_prompt"
USER_REPRESENTATION_METHOD: str = "long_term"
class HistorySettings(BaseSettings):
model_config = SettingsConfigDict(env_prefix='HISTORY_')
class HistorySettings(TomlSettings):
model_config = SettingsConfigDict(env_prefix="HISTORY_")
MESSAGES_PER_SHORT_SUMMARY: int = 20
MESSAGES_PER_LONG_SUMMARY: int = 60
class AppSettings(BaseSettings):
class AppSettings(TomlSettings):
# No env_prefix for app-level settings
model_config = SettingsConfigDict(env_prefix="")
# Application-wide settings
LOG_LEVEL: str = "INFO"
FASTAPI_HOST: str = "0.0.0.0"
@ -101,36 +233,6 @@ class AppSettings(BaseSettings):
DERIVER: DeriverSettings = DeriverSettings()
HISTORY: HistorySettings = HistorySettings()
# For loading from a TOML file in the future:
# model_config = SettingsConfigDict(env_file_encoding='utf-8', extra='ignore', toml_file='config.toml')
# Global settings instance
settings = AppSettings()
# Example for loading prompts from files (can be uncommented and adapted)
# def load_prompt_from_file(file_path: str, default_prompt: str = "") -> str:
# expanded_path = os.path.expanduser(file_path) # Handles ~ for home directory
# if not os.path.isabs(expanded_path):
# # Assuming prompts directory is relative to the project root or a known location
# # This might need adjustment based on your project structure
# base_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Project root
# expanded_path = os.path.join(base_dir, file_path)
# try:
# with open(expanded_path, 'r') as f:
# return f.read().strip()
# except FileNotFoundError:
# # You might want to log a warning here
# # logger.warning(f"Prompt file not found: {expanded_path}. Using default.")
# return default_prompt
# except Exception as e:
# # logger.error(f"Error loading prompt file {expanded_path}: {e}")
# return default_prompt
# # Example of loading a specific prompt if its file path is set
# if settings.LLM.DIALECTIC_SYSTEM_PROMPT_FILE:
# settings.LLM.DIALECTIC_SYSTEM_PROMPT = load_prompt_from_file(
# settings.LLM.DIALECTIC_SYSTEM_PROMPT_FILE,
# default_prompt="Default dialectic system prompt if file is missing." # Provide a fallback
# )
# # Repeat for other file-based prompts

69
src/schemas.py
View File

@ -23,15 +23,12 @@ class AppUpdate(AppBase):
class App(AppBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
name: str
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class UserBase(BaseModel):
@ -53,16 +50,13 @@ class UserUpdate(UserBase):
class User(UserBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
name: str
app_id: str
created_at: datetime.datetime
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class MessageBase(BaseModel):
@ -84,19 +78,16 @@ class MessageUpdate(MessageBase):
class Message(MessageBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
content: str
is_user: bool
session_id: str
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
app_id: str
user_id: str
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class SessionBase(BaseModel):
@ -117,17 +108,14 @@ class SessionUpdate(SessionBase):
class Session(SessionBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
is_active: bool
user_id: str
app_id: str
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class MetamessageBase(BaseModel):
@ -135,7 +123,7 @@ class MetamessageBase(BaseModel):
class MetamessageCreate(MetamessageBase):
label: Annotated[str, Field(min_length=1, max_length=50, alias='metamessage_type')]
label: Annotated[str, Field(min_length=1, max_length=50, alias="metamessage_type")]
content: Annotated[str, Field(min_length=0, max_length=50000)]
session_id: str | None = None
message_id: str | None = None
@ -145,7 +133,7 @@ class MetamessageCreate(MetamessageBase):
class MetamessageGet(MetamessageBase):
label: str | None = Field(default=None, alias='metamessage_type')
label: str | None = Field(default=None, alias="metamessage_type")
session_id: str | None = None
message_id: str | None = None
filter: dict | None = None
@ -156,21 +144,21 @@ class MetamessageGet(MetamessageBase):
class MetamessageUpdate(MetamessageBase):
session_id: str | None = None
message_id: str | None = None
label: str | None = Field(default=None, alias='metamessage_type')
label: str | None = Field(default=None, alias="metamessage_type")
metadata: dict | None = None
model_config = ConfigDict(populate_by_name=True)
class Metamessage(MetamessageBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
label: str
content: str
user_id: str
app_id: str
session_id: str | None
message_id: str | None
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
# Included for backwards compatibility with the old metamessage_type field
@ -179,10 +167,7 @@ class Metamessage(MetamessageBase):
def metamessage_type(self) -> str:
return self.label
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class CollectionBase(BaseModel):
@ -216,17 +201,14 @@ class CollectionUpdate(CollectionBase):
class Collection(CollectionBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
name: str
user_id: str
app_id: str
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class DocumentBase(BaseModel):
@ -254,18 +236,15 @@ class DocumentUpdate(DocumentBase):
class Document(DocumentBase):
public_id: str = Field(serialization_alias='id')
public_id: str = Field(serialization_alias="id")
content: str
h_metadata: dict = Field(default={}, serialization_alias='metadata')
h_metadata: dict = Field(default={}, serialization_alias="metadata")
created_at: datetime.datetime
collection_id: str
app_id: str
user_id: str
model_config = ConfigDict(
from_attributes=True,
populate_by_name=True
)
model_config = ConfigDict(from_attributes=True, populate_by_name=True)
class DialecticOptions(BaseModel):

2
src/utils/history.py
View File

@ -199,7 +199,7 @@ Provide a {"comprehensive" if summary_type == SummaryType.LONG else "concise"} s
# Create a model client
client = ModelClient(
provider=ModelProvider(settings.LLM.SUMMARY_PROVIDER),
model=settings.LLM.SUMMARY_MODEL
model=settings.LLM.SUMMARY_MODEL,
)
# Generate the summary

6
src/utils/model_client.py
View File

@ -16,6 +16,8 @@ from langfuse.decorators import langfuse_context, observe
# from openai import AsyncOpenAI
from langfuse.openai import AsyncOpenAI
from src.config import settings
# Load environment variables
load_dotenv()
@ -48,8 +50,8 @@ OPENAI_COMPATIBLE_PROVIDERS = [
ModelProvider.GROQ,
]
DEFAULT_TEMPERATURE = 0.0
DEFAULT_MAX_TOKENS = 1000
DEFAULT_TEMPERATURE: float = settings.LLM.DEFAULT_TEMPERATURE
DEFAULT_MAX_TOKENS: int = settings.LLM.DEFAULT_MAX_TOKENS
class Message(Protocol):

14
tests/conftest.py
View File

@ -21,6 +21,7 @@ from src.dependencies import get_db
from src.exceptions import HonchoException
from src.security import create_admin_jwt, create_jwt, JWTParams
from src.main import app
from src.config import settings
# Create a custom handler that doesn't get closed prematurely
@ -45,18 +46,17 @@ logging.getLogger("sqlalchemy.engine.Engine").disabled = True
# Test database URL
# TODO use environment variable
CONNECTION_URI = make_url(
os.getenv(
"CONNECTION_URI",
"postgresql+psycopg://postgres:postgres@localhost:5432/postgres",
)
DB_URI = (
settings.DB.CONNECTION_URI
or "postgresql+psycopg://postgres:postgres@localhost:5432/postgres"
)
CONNECTION_URI = make_url(DB_URI)
TEST_DB_URL = CONNECTION_URI.set(database="test_db")
DEFAULT_DB_URL = str(CONNECTION_URI.set(database="postgres"))
# Test API authorization
USE_AUTH = os.getenv("USE_AUTH", "False").lower() == "true"
AUTH_JWT_SECRET = os.getenv("AUTH_JWT_SECRET", "test-secret")
USE_AUTH = settings.AUTH.USE_AUTH
AUTH_JWT_SECRET = settings.AUTH.JWT_SECRET
def create_test_database(db_url):

92
tests/routes/test_messages.py
View File

@ -7,7 +7,9 @@ from src import models # Import your SQLAlchemy models
async def test_create_message(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
@ -31,11 +33,17 @@ async def test_create_message(client, db_session, sample_data):
async def test_get_messages(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.public_id, content="Test message", is_user=True, app_id=test_app.public_id, user_id=test_user.public_id
session_id=test_session.public_id,
content="Test message",
is_user=True,
app_id=test_app.public_id,
user_id=test_user.public_id,
)
db_session.add(test_message)
await db_session.commit()
@ -57,24 +65,28 @@ async def test_get_messages(client, db_session, sample_data):
async def test_messages_pagination(client, db_session, sample_data):
"""Test pagination of messages with different page sizes."""
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.public_id)
test_session = models.Session(
app_id=test_app.public_id, user_id=test_user.public_id
)
db_session.add(test_session)
await db_session.commit()
# Create 50 test messages
for i in range(50):
test_message = models.Message(
app_id=test_app.public_id,
user_id=test_user.public_id,
session_id=test_session.public_id,
content=f"Pagination test message {i}",
is_user=i % 2 == 0, # Alternating user/non-user messages
h_metadata={"index": i}
h_metadata={"index": i},
)
db_session.add(test_message)
await db_session.commit()
# Test case 1: Default pagination (page 1, default size)
response = client.post(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list",
@ -82,12 +94,12 @@ async def test_messages_pagination(client, db_session, sample_data):
)
assert response.status_code == 200
data = response.json()
# Check pagination metadata
assert data["page"] == 1
assert "total" in data
assert data["total"] == 50 # Total count should be 50
# Test case 2: 5 pages of 10 items each
for page in range(1, 6):
response = client.post(
@ -96,22 +108,22 @@ async def test_messages_pagination(client, db_session, sample_data):
)
assert response.status_code == 200
data = response.json()
# Check pagination metadata
assert data["page"] == page
assert data["size"] == 10
assert data["total"] == 50
# Check items count (should be 10 for all pages)
expected_items = 10
assert len(data["items"]) == expected_items
# Verify we have the correct page of items
start_idx = (page - 1) * 10
for i, item in enumerate(data["items"]):
expected_idx = start_idx + i
assert item["metadata"]["index"] == expected_idx
# Test case 3: 2 pages of 25 items each
for page in range(1, 3):
response = client.post(
@ -120,22 +132,22 @@ async def test_messages_pagination(client, db_session, sample_data):
)
assert response.status_code == 200
data = response.json()
# Check pagination metadata
assert data["page"] == page
assert data["size"] == 25
assert data["total"] == 50
# Check items count
expected_items = 25
assert len(data["items"]) == expected_items
# Verify we have the correct page of items
start_idx = (page - 1) * 25
for i, item in enumerate(data["items"]):
expected_idx = start_idx + i
assert item["metadata"]["index"] == expected_idx
# Test case 4: 1 page of 50 items
response = client.post(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list?page=1&size=50",
@ -143,19 +155,19 @@ async def test_messages_pagination(client, db_session, sample_data):
)
assert response.status_code == 200
data = response.json()
# Check pagination metadata
assert data["page"] == 1
assert data["size"] == 50
assert data["total"] == 50
# Check items count
assert len(data["items"]) == 50
# Verify all items are included
all_indices = {item["metadata"]["index"] for item in data["items"]}
assert all_indices == set(range(50))
# Test case 5: Test with reverse=true (newest first, reverse order)
response = client.post(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/sessions/{test_session.public_id}/messages/list?page=1&size=50&reverse=true",
@ -163,7 +175,7 @@ async def test_messages_pagination(client, db_session, sample_data):
)
assert response.status_code == 200
data = response.json()
# Check items are in reverse order when reverse=true
for i, item in enumerate(data["items"]):
expected_idx = 49 - i
@ -174,7 +186,9 @@ async def test_messages_pagination(client, db_session, sample_data):
async def test_get_filtered_messages(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
@ -215,11 +229,17 @@ async def test_get_filtered_messages(client, db_session, sample_data):
async def test_update_message(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.public_id, content="Test message", is_user=True, app_id=test_app.public_id, user_id=test_user.public_id
session_id=test_session.public_id,
content="Test message",
is_user=True,
app_id=test_app.public_id,
user_id=test_user.public_id,
)
db_session.add(test_message)
await db_session.commit()
@ -237,11 +257,17 @@ async def test_update_message(client, db_session, sample_data):
async def test_update_message_empty_metadata(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session and message
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
test_message = models.Message(
session_id=test_session.public_id, content="Test message", is_user=True, app_id=test_app.public_id, user_id=test_user.public_id
session_id=test_session.public_id,
content="Test message",
is_user=True,
app_id=test_app.public_id,
user_id=test_user.public_id,
)
db_session.add(test_message)
await db_session.commit()
@ -260,7 +286,9 @@ async def test_update_message_empty_metadata(client, db_session, sample_data):
async def test_create_batch_messages(client, db_session, sample_data):
test_app, test_user = sample_data
# Create a test session
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()
@ -308,7 +336,9 @@ async def test_create_batch_messages(client, db_session, sample_data):
@pytest.mark.asyncio
async def test_create_batch_messages_limit(client, db_session, sample_data):
test_app, test_user = sample_data
test_session = models.Session(user_id=test_user.public_id, app_id=test_app.public_id)
test_session = models.Session(
user_id=test_user.public_id, app_id=test_app.public_id
)
db_session.add(test_session)
await db_session.commit()

2228
uv.lock
View File

File diff suppressed because it is too large Load Diff