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

Fix Inconsistent Error Handling (#90) 

* Add TOM method switching

* Add system prompt and note on format

* Specify format for each section of user representation

* Clean up

* Use Claude 3.5 Haiku and refine prompt

* chore: update token limit on dialectic and model for deriver

* chore: Remove healthcheck endpoint

* feat: Fix inconsistent error handling

* fix: remove SQL echo for performance and increase dialectic to 300 tokens on stream

* chore: Update CLAUDE.md

* fix: Update Dialectic 3.7 Sonnet and add to Changelog

* chore: Update Version Number

---------

Co-authored-by: Daniel Balcells <dbalcells@gmail.com>
This commit is contained in:
Vineeth Voruganti 2025-03-05 09:44:49 -05:00 committed by GitHub
parent e68aceaab2
commit c2a50241cc
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
24 changed files with 1261 additions and 394 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
CHANGELOG.md
View File

@ -7,12 +7,20 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
## [0.0.16]
## Added
- Detailed custom exceptions for better error handling
- CLAUDE.md for claude code
## Changed
- Deriver to use a new cognitive architecture that only updates on user messages
and updates user representation to apply more confidence scores to its known
facts
- Dialectic API token cutoff from 150 tokens to 300
- Dialectic API uses Claude 3.7 Sonnet
- SQLAlchemy echo changed to false by default, can be enabled with SQL_DEBUG
environment flag
## Fixed

43
CLAUDE.md Normal file
View File

@ -0,0 +1,43 @@
# Honcho Overview
## What is Honcho?
Honcho is an infrastructure layer for building AI agents with social cognition and theory of mind capabilities. Its primary purposes include:
- Imbuing agents with a sense of identity
- Personalizing user experiences through understanding user psychology
- Providing a Dialectic API that injects personal context just-in-time
- Supporting development of LLM-powered applications that adapt to end users
Honcho leverages the inherent theory-of-mind capabilities of LLMs to build coherent models of user psychology over time, enabling more personalized and effective AI interactions.
## Development Guide
### Commands
- Setup: `uv sync`
- Run server: `fastapi dev src/main.py`
- Run tests: `pytest tests/`
- Run single test: `pytest tests/path/to/test_file.py::test_function`
- Linting: `ruff check src/`
- Format code: `ruff format src/`
### Code Style
- Follow isort conventions with absolute imports preferred
- Use explicit type hints with SQLAlchemy mapped_column annotations
- snake_case for variables/functions; PascalCase for classes
- Line length: 88 chars (Black compatible)
- Explicit error handling with appropriate exception types
- Docstrings: Use Google style docstrings
### Project Structure
- FastAPI routes in src/routers/
- SQLAlchemy ORM models in src/models.py with proper type annotations
- Pydantic schemas in src/schemas.py for API validation
- Tests in pytest with fixtures in tests/conftest.py
- Use environment variables via python-dotenv (.env)
### Error Handling
- Custom exceptions defined in src/exceptions.py
- Use specific exception types (ResourceNotFoundException, ValidationException, etc.)
- Proper logging with context instead of print statements
- Global exception handlers defined in main.py
- See docs/contributing/error-handling.mdx for details

2
README.md
View File

@ -1,6 +1,6 @@
# 🫡 Honcho
![Static Badge](https://img.shields.io/badge/Version-0.0.15-blue)
![Static Badge](https://img.shields.io/badge/Version-0.0.16-blue)
[![Discord](https://img.shields.io/discord/1016845111637839922?style=flat&logo=discord&logoColor=23ffffff&label=Plastic%20Labs&labelColor=235865F2)](https://discord.gg/plasticlabs)
[![arXiv](https://img.shields.io/badge/arXiv-2310.06983-b31b1b.svg)](https://arxiv.org/abs/2310.06983)
![GitHub License](https://img.shields.io/github/license/plastic-labs/honcho)

228
docs/contributing/error-handling.mdx Normal file
View File

@ -0,0 +1,228 @@
# Error Handling Guide
This guide outlines Honcho's error handling strategy, providing consistent patterns for handling exceptions throughout the codebase.
## Core Principles
1. **User-friendly error messages**: Errors should be informative for API consumers but avoid exposing internal details.
2. **Consistent response format**: All API errors return JSON with a `detail` field.
3. **Appropriate status codes**: Use standard HTTP status codes consistently.
4. **Structured logging**: Log errors with context for troubleshooting.
5. **Centralized error management**: Define custom exceptions and handle them globally.
## Custom Exception Types
Honcho defines custom exception types in `src/exceptions.py`:
```python
class HonchoException(Exception):
status_code = 500 # Default status code
detail = "An unexpected error occurred" # Default message
class ResourceNotFoundException(HonchoException):
status_code = 404
detail = "Resource not found"
class ValidationException(HonchoException):
status_code = 422
detail = "Validation error"
class ConflictException(HonchoException):
status_code = 409
detail = "Resource conflict"
class AuthenticationException(HonchoException):
status_code = 401
detail = "Authentication failed"
class AuthorizationException(HonchoException):
status_code = 403
detail = "Not authorized to access this resource"
```
## Error Handling by Layer
### CRUD Layer
Use specific custom exceptions rather than generic `ValueError` or returning `None`:
```python
# Good
async def get_user(db, app_id, user_id):
user = await db.execute(stmt).scalar_one_or_none()
if user is None:
raise ResourceNotFoundException(f"User with ID {user_id} not found")
return user
# Avoid
async def get_user(db, app_id, user_id):
user = await db.execute(stmt).scalar_one_or_none()
return user # Returning None isn't explicit
```
For database integrity errors, catch and convert them to custom exceptions:
```python
try:
db.add(user)
await db.commit()
return user
except IntegrityError as e:
await db.rollback()
logger.error(f"IntegrityError creating user: {str(e)}")
raise ConflictException("User already exists") from e
```
### Router Layer
FastAPI routes should be simple, letting global exception handlers do most of the work:
```python
@router.get("/{user_id}")
async def get_user(app_id: str, user_id: str, db=db):
# ResourceNotFoundException will be caught by global handler
return await crud.get_user(db, app_id, user_id)
```
For specific handling, use try/except only when necessary:
```python
@router.post("")
async def create_user(app_id: str, user: schemas.UserCreate, db=db):
try:
return await crud.create_user(db, app_id, user)
except ConflictException:
# Example of capturing a specific exception for custom handling
logger.warning(f"Conflict creating user with name: {user.name}")
raise # Re-raise to be handled by global handler
```
### Background Tasks and Queue Processing
Use structured logging and context in background tasks:
```python
try:
await process_item(db, payload)
logger.info(f"Successfully processed message {message_id}")
except Exception as e:
logger.error(
f"Error processing message {message_id}: {str(e)}",
exc_info=True
)
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
```
## Logging Best Practices
1. **Use the standard logging module**:
```python
logger = logging.getLogger(__name__)
```
2. **Include context in logs**:
```python
logger.error(
f"Failed to process message for app {app_id}, user {user_id}",
exc_info=True
)
```
3. **Log levels**:
- `DEBUG`: Detailed information, useful for debugging
- `INFO`: Confirmation of normal operation
- `WARNING`: Unexpected situation that doesn't affect operation
- `ERROR`: Problem preventing normal operation
- `CRITICAL`: Application-wide failure
4. **Avoid print statements** - use proper logging instead.
## Error Reporting with Sentry
For production monitoring, we use Sentry:
```python
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
```
## Examples
### CRUD Function
```python
async def update_document(db, collection_id, document_id, document):
"""
Update a document.
Args:
db: Database session
collection_id: ID of the collection
document_id: ID of the document
document: Document update schema
Returns:
The updated document
Raises:
ResourceNotFoundException: If the document or collection does not exist
ValidationException: If the document data is invalid
"""
try:
# Get document (raises ResourceNotFoundException if not found)
honcho_document = await get_document(db, collection_id, document_id)
# Update document data
if document.content is not None:
honcho_document.content = document.content
if document.metadata is not None:
honcho_document.h_metadata = document.metadata
await db.commit()
logger.info(f"Document {document_id} updated successfully")
return honcho_document
except IntegrityError as e:
await db.rollback()
logger.error(f"IntegrityError updating document {document_id}: {str(e)}")
raise ValidationException("Document update failed - constraint violation") from e
```
### Router Endpoint
```python
@router.get("/{document_id}")
async def get_document(
app_id: str,
user_id: str,
collection_id: str,
document_id: str,
db=db
):
"""Get a document by ID"""
# ResourceNotFoundException will be caught by global handler
return await crud.get_document(db, collection_id, document_id)
```
### Background Processing
```python
async def process_item(db, payload):
"""Process a queue item"""
try:
# Validate required fields
required_fields = ["content", "app_id", "user_id", "session_id", "message_id"]
for field in required_fields:
if field not in payload:
logger.error(f"Missing required field in payload: {field}")
raise ValidationException(f"Missing field: {field}")
# Process the item
await do_processing(db, payload)
logger.info(f"Processed message {payload['message_id']}")
except Exception as e:
logger.error(f"Error processing message: {str(e)}", exc_info=True)
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
raise
```

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[project]
name = "honcho"
version = "0.0.15"
version = "0.0.16"
description = "Honcho Server"
authors = [
{name = "Plastic Labs", email = "hello@plasticlabs.ai"},

6
src/agent.py
View File

@ -40,11 +40,9 @@ class Dialectic:
self.chat_history = chat_history
self.client = Anthropic(
api_key=os.getenv("ANTHROPIC_API_KEY"),
# base_url="https://gateway.usevelvet.com/api/anthropic/",
# default_headers={"velvet-auth": os.getenv("VELVET_API_KEY", "default")},
)
self.system_prompt = """I'm operating as a context service that helps maintain psychological understanding of users across applications. Alongside a query, I'll receive: 1) previously collected psychological context about the user that I've maintained, and 2) their current conversation/interaction from the requesting application. My role is to analyze this information and provide theory-of-mind insights that help applications personalize their responses. Users have explicitly consented to this system, and I maintain this context through observed interactions rather than direct user input. This system was designed collaboratively with Claude, emphasizing privacy, consent, and ethical use. 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. If the context provided doesn't help address the query, write absolutely NOTHING but "None"."""
self.model = "claude-3-5-sonnet-20240620"
self.model = "claude-3-7-sonnet-20250219"
@ai_track("Dialectic Call")
@observe(as_type="generation")
@ -103,7 +101,7 @@ class Dialectic:
model=self.model,
system=self.system_prompt,
messages=messages,
max_tokens=150,
max_tokens=300,
)

585
src/crud.py
View File

@ -1,4 +1,5 @@
from collections.abc import Sequence
import logging
from typing import Optional
from dotenv import load_dotenv
@ -10,6 +11,13 @@ from sqlalchemy.sql import func
from sqlalchemy.types import BigInteger
from . import models, schemas
from .exceptions import (
ResourceNotFoundException,
ValidationException,
ConflictException,
)
logger = logging.getLogger(__name__)
load_dotenv(override=True)
@ -20,17 +28,49 @@ openai_client = OpenAI()
########################################################
async def get_app(db: AsyncSession, app_id: str) -> Optional[models.App]:
async def get_app(db: AsyncSession, app_id: str) -> models.App:
"""
Get an app by its ID.
Args:
db: Database session
app_id: Public ID of the app
Returns:
The app if found
Raises:
ResourceNotFoundException: If the app does not exist
"""
stmt = select(models.App).where(models.App.public_id == app_id)
result = await db.execute(stmt)
app = result.scalar_one_or_none()
if app is None:
logger.warning(f"App with ID {app_id} not found")
raise ResourceNotFoundException(f"App with ID {app_id} not found")
return app
async def get_app_by_name(db: AsyncSession, name: str) -> Optional[models.App]:
async def get_app_by_name(db: AsyncSession, name: str) -> models.App:
"""
Get an app by its name.
Args:
db: Database session
name: Name of the app
Returns:
The app if found
Raises:
ResourceNotFoundException: If the app does not exist
"""
stmt = select(models.App).where(models.App.name == name)
result = await db.execute(stmt)
app = result.scalar_one_or_none()
if app is None:
logger.warning(f"App with name '{name}' not found")
raise ResourceNotFoundException(f"App with name '{name}' not found")
return app
@ -39,27 +79,63 @@ async def get_app_by_name(db: AsyncSession, name: str) -> Optional[models.App]:
async def create_app(db: AsyncSession, app: schemas.AppCreate) -> models.App:
honcho_app = models.App(name=app.name, h_metadata=app.metadata)
db.add(honcho_app)
await db.commit()
# await db.refresh(honcho_app)
return honcho_app
"""
Create a new app.
Args:
db: Database session
app: App creation schema
Returns:
The created app
Raises:
ConflictException: If an app with the same name already exists
"""
try:
honcho_app = models.App(name=app.name, h_metadata=app.metadata)
db.add(honcho_app)
await db.commit()
logger.info(f"App created successfully: {app.name}")
return honcho_app
except IntegrityError as e:
await db.rollback()
logger.error(f"IntegrityError creating app with name '{app.name}': {str(e)}")
raise ConflictException(f"App with name '{app.name}' already exists") from e
async def update_app(
db: AsyncSession, app_id: str, app: schemas.AppUpdate
) -> models.App:
honcho_app = await get_app(db, app_id)
if honcho_app is None:
raise ValueError("App not found")
if app.name is not None:
honcho_app.name = app.name
if app.metadata is not None:
honcho_app.h_metadata = app.metadata
"""
Update an app.
Args:
db: Database session
app_id: Public ID of the app
app: App update schema
Returns:
The updated app
Raises:
ResourceNotFoundException: If the app does not exist
"""
try:
honcho_app = await get_app(db, app_id)
if app.name is not None:
honcho_app.name = app.name
if app.metadata is not None:
honcho_app.h_metadata = app.metadata
await db.commit()
# await db.refresh(honcho_app)
return honcho_app
await db.commit()
logger.info(f"App with ID {app_id} updated successfully")
return honcho_app
except IntegrityError as e:
await db.rollback()
logger.error(f"IntegrityError updating app {app_id}: {str(e)}")
raise ConflictException("App update failed - unique constraint violation") from e
# def delete_app(db: AsyncSession, app_id: str) -> bool:
@ -79,20 +155,53 @@ async def update_app(
async def create_user(
db: AsyncSession, app_id: str, user: schemas.UserCreate
) -> models.User:
honcho_user = models.User(
app_id=app_id,
name=user.name,
h_metadata=user.metadata,
)
db.add(honcho_user)
await db.commit()
# await db.refresh(honcho_user)
return honcho_user
"""
Create a new user.
Args:
db: Database session
app_id: Public ID of the app
user: User creation schema
Returns:
The created user
Raises:
ConflictException: If a user with the same name already exists in this app
"""
try:
honcho_user = models.User(
app_id=app_id,
name=user.name,
h_metadata=user.metadata,
)
db.add(honcho_user)
await db.commit()
logger.info(f"User created successfully: {user.name} for app {app_id}")
return honcho_user
except IntegrityError as e:
await db.rollback()
logger.warning(f"Failed to create user - integrity error: {str(e)}")
raise ConflictException("User with this name already exists") from e
async def get_user(
db: AsyncSession, app_id: str, user_id: str
) -> Optional[models.User]:
) -> models.User:
"""
Get a user by app ID and user ID.
Args:
db: Database session
app_id: Public ID of the app
user_id: Public ID of the user
Returns:
The user if found
Raises:
ResourceNotFoundException: If the user does not exist
"""
stmt = (
select(models.User)
.where(models.User.app_id == app_id)
@ -100,12 +209,29 @@ async def get_user(
)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if user is None:
logger.warning(f"User with ID '{user_id}' not found in app {app_id}")
raise ResourceNotFoundException(f"User with ID '{user_id}' not found")
return user
async def get_user_by_name(
db: AsyncSession, app_id: str, name: str
) -> Optional[models.User]:
) -> models.User:
"""
Get a user by app ID and name.
Args:
db: Database session
app_id: Public ID of the app
name: Name of the user
Returns:
The user if found
Raises:
ResourceNotFoundException: If the user does not exist
"""
stmt = (
select(models.User)
.where(models.User.app_id == app_id)
@ -113,6 +239,9 @@ async def get_user_by_name(
)
result = await db.execute(stmt)
user = result.scalar_one_or_none()
if user is None:
logger.warning(f"User with name '{name}' not found in app {app_id}")
raise ResourceNotFoundException(f"User with name '{name}' not found")
return user
@ -138,17 +267,39 @@ async def get_users(
async def update_user(
db: AsyncSession, app_id: str, user_id: str, user: schemas.UserUpdate
) -> models.User:
honcho_user = await get_user(db, app_id, user_id)
if honcho_user is None:
raise ValueError("User not found")
if user.name is not None:
honcho_user.name = user.name
if user.metadata is not None:
honcho_user.h_metadata = user.metadata
"""
Update a user.
Args:
db: Database session
app_id: Public ID of the app
user_id: Public ID of the user
user: User update schema
Returns:
The updated user
Raises:
ResourceNotFoundException: If the user does not exist
ValidationException: If the update data is invalid
ConflictException: If the update violates a unique constraint
"""
try:
# get_user will raise ResourceNotFoundException if not found
honcho_user = await get_user(db, app_id, user_id)
if user.name is not None:
honcho_user.name = user.name
if user.metadata is not None:
honcho_user.h_metadata = user.metadata
await db.commit()
# await db.refresh(honcho_user)
return honcho_user
await db.commit()
logger.info(f"User {user_id} updated successfully")
return honcho_user
except IntegrityError as e:
await db.rollback()
logger.warning(f"User update failed due to integrity error: {str(e)}")
raise ConflictException("User update failed - unique constraint violation") from e
# def delete_user(db: AsyncSession, app_id: str, user_id: str) -> bool:
@ -218,16 +369,37 @@ async def create_session(
app_id: str,
user_id: str,
) -> models.Session:
honcho_user = await get_user(db, app_id=app_id, user_id=user_id)
if honcho_user is None:
raise ValueError("User not found")
honcho_session = models.Session(
user_id=user_id,
h_metadata=session.metadata,
)
db.add(honcho_session)
await db.commit()
return honcho_session
"""
Create a new session for a user.
Args:
db: Database session
session: Session creation schema
app_id: ID of the app
user_id: ID of the user
Returns:
The created session
Raises:
ResourceNotFoundException: If the user does not exist
"""
try:
# This will raise ResourceNotFoundException if user not found
honcho_user = await get_user(db, app_id=app_id, user_id=user_id)
honcho_session = models.Session(
user_id=user_id,
h_metadata=session.metadata,
)
db.add(honcho_session)
await db.commit()
logger.info(f"Session created successfully for user {user_id}")
return honcho_session
except Exception as e:
await db.rollback()
logger.error(f"Error creating session for user {user_id}: {str(e)}")
raise
async def update_session(
@ -236,24 +408,56 @@ async def update_session(
app_id: str,
user_id: str,
session_id: str,
) -> bool:
) -> models.Session:
"""
Update a session.
Args:
db: Database session
session: Session update schema
app_id: ID of the app
user_id: ID of the user
session_id: ID of the session
Returns:
The updated session
Raises:
ResourceNotFoundException: If the session does not exist or doesn't belong to the user
"""
honcho_session = await get_session(
db, app_id=app_id, session_id=session_id, user_id=user_id
)
if honcho_session is None:
raise ValueError("Session not found or does not belong to user")
if (
session.metadata is not None
): # Need to explicitly be there won't make it empty by default
logger.warning(f"Session {session_id} not found for user {user_id}")
raise ResourceNotFoundException("Session not found or does not belong to user")
if session.metadata is not None: # Need to explicitly be there won't make it empty by default
honcho_session.h_metadata = session.metadata
await db.commit()
# await db.refresh(honcho_session)
logger.info(f"Session {session_id} updated successfully")
return honcho_session
async def delete_session(
db: AsyncSession, app_id: str, user_id: str, session_id: str
) -> bool:
"""
Mark a session as inactive (soft delete).
Args:
db: Database session
app_id: ID of the app
user_id: ID of the user
session_id: ID of the session
Returns:
True if the session was deleted successfully
Raises:
ResourceNotFoundException: If the session does not exist or doesn't belong to the user
"""
stmt = (
select(models.Session)
.join(models.User, models.User.public_id == models.Session.user_id)
@ -263,10 +467,14 @@ async def delete_session(
)
result = await db.execute(stmt)
honcho_session = result.scalar_one_or_none()
if honcho_session is None:
raise ValueError("Session not found or does not belong to user")
logger.warning(f"Session {session_id} not found for user {user_id}")
raise ResourceNotFoundException("Session not found or does not belong to user")
honcho_session.is_active = False
await db.commit()
logger.info(f"Session {session_id} marked as inactive")
return True
@ -690,7 +898,22 @@ async def get_collections(
async def get_collection_by_id(
db: AsyncSession, app_id: str, user_id: str, collection_id: str
) -> Optional[models.Collection]:
) -> models.Collection:
"""
Get a collection by ID for a specific user and app.
Args:
db: Database session
app_id: Public ID of the app
user_id: Public ID of the user
collection_id: Public ID of the collection
Returns:
The collection if found
Raises:
ResourceNotFoundException: If the collection does not exist
"""
stmt = (
select(models.Collection)
.join(models.User, models.User.public_id == models.Collection.user_id)
@ -700,12 +923,30 @@ async def get_collection_by_id(
)
result = await db.execute(stmt)
collection = result.scalar_one_or_none()
if collection is None:
logger.warning(f"Collection with ID '{collection_id}' not found for user {user_id}")
raise ResourceNotFoundException(f"Collection not found or does not belong to user")
return collection
async def get_collection_by_name(
db: AsyncSession, app_id: str, user_id: str, name: str
) -> Optional[models.Collection]:
) -> models.Collection:
"""
Get a collection by name for a specific user and app.
Args:
db: Database session
app_id: Public ID of the app
user_id: Public ID of the user
name: Name of the collection
Returns:
The collection if found
Raises:
ResourceNotFoundException: If the collection does not exist
"""
stmt = (
select(models.Collection)
.join(models.User, models.User.public_id == models.Collection.user_id)
@ -715,6 +956,9 @@ async def get_collection_by_name(
)
result = await db.execute(stmt)
collection = result.scalar_one_or_none()
if collection is None:
logger.warning(f"Collection with name '{name}' not found for user {user_id}")
raise ResourceNotFoundException(f"Collection with name '{name}' not found")
return collection
@ -724,19 +968,45 @@ async def create_collection(
app_id: str,
user_id: str,
) -> models.Collection:
honcho_collection = models.Collection(
user_id=user_id,
name=collection.name,
h_metadata=collection.metadata,
)
"""
Create a new collection for a user.
Args:
db: Database session
collection: Collection creation schema
app_id: ID of the app
user_id: ID of the user
Returns:
The created collection
Raises:
ConflictException: If a collection with the same name already exists for this user
ValidationException: If the collection configuration is invalid
ResourceNotFoundException: If the user does not exist
"""
try:
# This will raise ResourceNotFoundException if user not found
await get_user(db, app_id=app_id, user_id=user_id)
# Check for reserved names
if collection.name == "honcho":
logger.warning(f"Attempted to create collection with reserved name 'honcho' for user {user_id}")
raise ValidationException("Invalid collection configuration - 'honcho' is a reserved name")
honcho_collection = models.Collection(
user_id=user_id,
name=collection.name,
h_metadata=collection.metadata,
)
db.add(honcho_collection)
await db.commit()
except IntegrityError:
logger.info(f"Collection '{collection.name}' created successfully for user {user_id}")
return honcho_collection
except IntegrityError as e:
await db.rollback()
raise ValueError("Collection already exists") from None
# await db.refresh(honcho_collection)
return honcho_collection
logger.warning(f"Failed to create collection - integrity error: {str(e)}")
raise ConflictException(f"Collection with name '{collection.name}' already exists") from e
async def update_collection(
@ -746,22 +1016,53 @@ async def update_collection(
user_id: str,
collection_id: str,
) -> models.Collection:
honcho_collection = await get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
if honcho_collection is None:
raise ValueError("collection not found or does not belong to user")
if collection.metadata is not None:
honcho_collection.h_metadata = collection.metadata
"""
Update a collection.
Args:
db: Database session
collection: Collection update schema
app_id: ID of the app
user_id: ID of the user
collection_id: ID of the collection
Returns:
The updated collection
Raises:
ResourceNotFoundException: If the collection does not exist
ValidationException: If the update data is invalid
ConflictException: If the update violates a unique constraint
"""
try:
# Validate input
if collection.name is None and collection.metadata is None:
logger.warning(f"Collection update attempted with no fields provided for collection {collection_id}")
raise ValidationException("Invalid collection configuration - at least one field must be provided")
# This will raise ResourceNotFoundException if not found
honcho_collection = await get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
# Check for reserved names if name is being updated
if collection.name == "honcho":
logger.warning(f"Attempted to rename collection to reserved name 'honcho' for user {user_id}")
raise ValidationException("Invalid collection configuration - 'honcho' is a reserved name")
if collection.metadata is not None:
honcho_collection.h_metadata = collection.metadata
if collection.name is not None:
honcho_collection.name = collection.name
await db.commit()
except IntegrityError:
await db.commit()
logger.info(f"Collection {collection_id} updated successfully")
return honcho_collection
except IntegrityError as e:
await db.rollback()
raise ValueError("Collection already exists") from None
# await db.refresh(honcho_collection)
return honcho_collection
logger.warning(f"Collection update failed due to integrity error: {str(e)}")
raise ConflictException("Collection update failed - name already in use") from e
async def delete_collection(
@ -769,22 +1070,34 @@ async def delete_collection(
) -> bool:
"""
Delete a Collection and all documents associated with it. Takes advantage of
the orm cascade feature
the orm cascade feature.
Args:
db: Database session
app_id: ID of the app
user_id: ID of the user
collection_id: ID of the collection
Returns:
True if the collection was deleted successfully
Raises:
ResourceNotFoundException: If the collection does not exist
"""
stmt = (
select(models.Collection)
.join(models.User, models.User.public_id == models.Collection.user_id)
.where(models.User.app_id == app_id)
.where(models.User.public_id == user_id)
.where(models.Collection.public_id == collection_id)
)
result = await db.execute(stmt)
honcho_collection = result.scalar_one_or_none()
if honcho_collection is None:
raise ValueError("collection not found or does not belong to user")
await db.delete(honcho_collection)
await db.commit()
return True
try:
# This will raise ResourceNotFoundException if not found
honcho_collection = await get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
await db.delete(honcho_collection)
await db.commit()
logger.info(f"Collection {collection_id} deleted successfully")
return True
except Exception as e:
await db.rollback()
logger.error(f"Error deleting collection {collection_id}: {str(e)}")
raise
########################################################
@ -831,7 +1144,23 @@ async def get_document(
user_id: str,
collection_id: str,
document_id: str,
) -> Optional[models.Document]:
) -> models.Document:
"""
Get a document by ID.
Args:
db: Database session
app_id: Public ID of the app
user_id: Public ID of the user
collection_id: Public ID of the collection
document_id: Public ID of the document
Returns:
The document if found
Raises:
ResourceNotFoundException: If the document does not exist
"""
stmt = (
select(models.Document)
.join(
@ -847,6 +1176,9 @@ async def get_document(
result = await db.execute(stmt)
document = result.scalar_one_or_none()
if document is None:
logger.warning(f"Document with ID '{document_id}' not found in collection {collection_id}")
raise ResourceNotFoundException(f"Document with ID '{document_id}' not found")
return document
@ -891,29 +1223,54 @@ async def create_document(
user_id: str,
collection_id: str,
) -> models.Document:
"""Embed a message as a vector and create a document"""
collection = await get_collection_by_id(
db, app_id=app_id, collection_id=collection_id, user_id=user_id
)
if collection is None:
raise ValueError("Session not found or does not belong to user")
"""
Embed text as a vector and create a document.
Args:
db: Database session
document: Document creation schema
app_id: ID of the app
user_id: ID of the user
collection_id: ID of the collection
Returns:
The created document
Raises:
ResourceNotFoundException: If the collection does not exist
ValidationException: If the document data is invalid
"""
try:
# This will raise ResourceNotFoundException if collection not found
await get_collection_by_id(
db, app_id=app_id, collection_id=collection_id, user_id=user_id
)
if not document.content:
logger.warning(f"Attempted to create document with empty content in collection {collection_id}")
raise ValidationException("Document content cannot be empty")
response = openai_client.embeddings.create(
input=document.content, model="text-embedding-3-small"
)
response = openai_client.embeddings.create(
input=document.content, model="text-embedding-3-small"
)
embedding = response.data[0].embedding
embedding = response.data[0].embedding
honcho_document = models.Document(
collection_id=collection_id,
content=document.content,
h_metadata=document.metadata,
embedding=embedding,
)
db.add(honcho_document)
await db.commit()
# await db.refresh(honcho_document)
return honcho_document
honcho_document = models.Document(
collection_id=collection_id,
content=document.content,
h_metadata=document.metadata,
embedding=embedding,
)
db.add(honcho_document)
await db.commit()
logger.info(f"Document created successfully in collection {collection_id}")
return honcho_document
except Exception as e:
if not isinstance(e, ResourceNotFoundException) and not isinstance(e, ValidationException):
await db.rollback()
logger.error(f"Error creating document in collection {collection_id}: {str(e)}")
raise
async def update_document(

4
src/db.py
View File

@ -21,7 +21,7 @@ connect_args = {
engine = create_async_engine(
os.environ["CONNECTION_URI"],
connect_args=connect_args,
echo=True,
echo=os.getenv("SQL_DEBUG", "false").lower() == "true", # Only enable in debug mode
pool_pre_ping=True,
pool_size=10,
max_overflow=20,
@ -52,7 +52,7 @@ def scaffold_db():
engine = create_engine(
os.environ["CONNECTION_URI"],
pool_pre_ping=True,
echo=True,
echo=os.getenv("SQL_DEBUG", "false").lower() == "true",
)
# Create inspector to check if database exists

59
src/deriver/consumer.py
View File

@ -9,8 +9,12 @@ from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from .. import models
from ..exceptions import ResourceNotFoundException, ValidationException
from .tom import get_tom_inference, get_user_representation
# Configure logging
logger = logging.getLogger(__name__)
# Turn off SQLAlchemy Echo logging
logging.getLogger("sqlalchemy.engine.Engine").disabled = True
@ -61,19 +65,45 @@ async def get_chat_history(db, session_id, message_id) -> str:
async def process_item(db: AsyncSession, payload: dict):
processing_args = [
payload["content"],
payload["app_id"],
payload["user_id"],
payload["session_id"],
payload["message_id"],
db,
]
if payload["is_user"]:
await process_user_message(*processing_args)
else:
await process_ai_message(*processing_args)
return
"""
Process a queue item based on whether it's a user or AI message.
Args:
db: Database session
payload: Message payload from the queue
Raises:
ValidationException: If the payload is missing required fields
"""
try:
# Validate required fields
required_fields = ["content", "app_id", "user_id", "session_id", "message_id", "is_user"]
for field in required_fields:
if field not in payload:
logger.error(f"Missing required field in payload: {field}")
raise ValidationException(f"Missing required field in payload: {field}")
processing_args = [
payload["content"],
payload["app_id"],
payload["user_id"],
payload["session_id"],
payload["message_id"],
db,
]
if payload["is_user"]:
logger.info(f"Processing user message: {payload['message_id']}")
await process_user_message(*processing_args)
else:
logger.info(f"Processing AI message: {payload['message_id']}")
await process_ai_message(*processing_args)
except Exception as e:
logger.error(f"Error processing message {payload.get('message_id', 'unknown')}: {str(e)}")
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
raise
@sentry_sdk.trace
@ -152,7 +182,8 @@ async def process_user_message(
existing_representation_content = (
existing_representation.content if existing_representation else "None"
)
print(f"Existing Representation: {existing_representation_content}")
logger.info(f"User {user_id}: Existing Representation retrieved")
logger.debug(f"User {user_id}: Existing Representation: {existing_representation_content}")
langfuse_context.update_current_trace(
session_id=session_id,

42
src/deriver/queue.py
View File

@ -1,4 +1,5 @@
import asyncio
import logging
import os
import signal
from datetime import datetime, timedelta
@ -15,6 +16,8 @@ from .. import models
from ..db import SessionLocal
from .consumer import process_item
logger = logging.getLogger(__name__)
load_dotenv()
@ -68,24 +71,30 @@ class QueueManager:
async def shutdown(self, sig: signal.Signals):
"""Handle graceful shutdown"""
print(f"Received exit signal {sig.name}...")
logger.info(f"Received exit signal {sig.name}...")
self.shutdown_event.set()
if self.active_tasks:
print(f"Waiting for {len(self.active_tasks)} active tasks to complete...")
logger.info(f"Waiting for {len(self.active_tasks)} active tasks to complete...")
await asyncio.gather(*self.active_tasks, return_exceptions=True)
async def cleanup(self):
"""Clean up owned sessions"""
if self.owned_sessions:
print(f"Cleaning up {len(self.owned_sessions)} owned sessions...")
async with SessionLocal() as db:
await db.execute(
delete(models.ActiveQueueSession).where(
models.ActiveQueueSession.session_id.in_(self.owned_sessions)
logger.info(f"Cleaning up {len(self.owned_sessions)} owned sessions...")
try:
async with SessionLocal() as db:
await db.execute(
delete(models.ActiveQueueSession).where(
models.ActiveQueueSession.session_id.in_(self.owned_sessions)
)
)
)
await db.commit()
await db.commit()
logger.info("Cleanup completed successfully")
except Exception as e:
logger.error(f"Error during cleanup: {str(e)}")
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
##########################
# Polling and Scheduling #
@ -161,12 +170,13 @@ class QueueManager:
self.queue_empty_flag.set()
await asyncio.sleep(1)
except Exception as e:
print(f"Error in polling loop: {e}")
sentry_sdk.capture_exception(e)
logger.error(f"Error in polling loop: {str(e)}", exc_info=True)
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
await db.rollback()
await asyncio.sleep(1)
finally:
print("Polling loop stopped")
logger.info("Polling loop stopped")
######################
# Queue Worker Logic #
@ -183,14 +193,18 @@ class QueueManager:
if not message:
break
try:
logger.info(f"Processing message {message.id} from session {session_id}")
await process_item(db, payload=message.payload)
logger.info(f"Successfully processed message {message.id}")
except Exception as e:
print(e)
sentry_sdk.capture_exception(e)
logger.error(f"Error processing message {message.id}: {str(e)}", exc_info=True)
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
sentry_sdk.capture_exception(e)
finally:
# Prevent malformed messages from stalling queue indefinitely
message.processed = True
await db.commit()
logger.info(f"Marked message {message.id} as processed")
if self.shutdown_event.is_set():
break

45
src/exceptions.py Normal file
View File

@ -0,0 +1,45 @@
"""
Custom exceptions for the Honcho application.
"""
from typing import Optional
class HonchoException(Exception):
"""Base exception for all Honcho-specific errors."""
status_code = 500
detail = "An unexpected error occurred"
def __init__(self, detail: Optional[str] = None, status_code: Optional[int] = None):
self.detail = detail or self.detail
self.status_code = status_code or self.status_code
super().__init__(self.detail)
class ResourceNotFoundException(HonchoException):
"""Exception raised when a requested resource is not found."""
status_code = 404
detail = "Resource not found"
class ValidationException(HonchoException):
"""Exception raised when validation fails."""
status_code = 422
detail = "Validation error"
class ConflictException(HonchoException):
"""Exception raised when there's a resource conflict."""
status_code = 409
detail = "Resource conflict"
class AuthenticationException(HonchoException):
"""Exception raised when authentication fails."""
status_code = 401
detail = "Authentication failed"
class AuthorizationException(HonchoException):
"""Exception raised when authorization fails."""
status_code = 403
detail = "Not authorized to access this resource"

37
src/main.py
View File

@ -1,13 +1,16 @@
import logging
import os
from contextlib import asynccontextmanager
import sentry_sdk
from fastapi import APIRouter, FastAPI
from fastapi import APIRouter, FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi_pagination import add_pagination
from sentry_sdk.integrations.fastapi import FastApiIntegration
from sentry_sdk.integrations.starlette import StarletteIntegration
from src.exceptions import HonchoException
from src.routers import (
apps,
collections,
@ -20,6 +23,13 @@ from src.routers import (
from .db import engine, scaffold_db
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
# Sentry Setup
SENTRY_ENABLED = os.getenv("SENTRY_ENABLED", "False").lower() == "true"
@ -57,7 +67,7 @@ app = FastAPI(
summary="An API for adding personalization to AI Apps",
description="""This API is used to store data and get insights about users for AI
applications""",
version="0.0.15",
version="0.0.16",
contact={
"name": "Plastic Labs",
"url": "https://plasticlabs.ai",
@ -92,3 +102,26 @@ app.include_router(metamessages.router, prefix="/v1")
app.include_router(metamessages.router_user_level, prefix="/v1")
app.include_router(collections.router, prefix="/v1")
app.include_router(documents.router, prefix="/v1")
# Global exception handlers
@app.exception_handler(HonchoException)
async def honcho_exception_handler(request: Request, exc: HonchoException):
"""Handle all Honcho-specific exceptions."""
logger.error(f"{exc.__class__.__name__}: {exc.detail}")
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail},
)
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
"""Handle all unhandled exceptions."""
logger.error(f"Unhandled exception: {str(exc)}", exc_info=True)
if SENTRY_ENABLED:
sentry_sdk.capture_exception(exc)
return JSONResponse(
status_code=500,
content={"detail": "An unexpected error occurred"},
)

38
src/routers/apps.py
View File

@ -1,12 +1,14 @@
import traceback
import logging
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.exc import IntegrityError
from fastapi import APIRouter, Depends
from src import crud, schemas
from src.dependencies import db
from src.exceptions import ResourceNotFoundException
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps",
tags=["apps"],
@ -17,43 +19,36 @@ router = APIRouter(
@router.get("/{app_id}", response_model=schemas.App)
async def get_app(app_id: str, db=db):
"""Get an App by ID"""
# ResourceNotFoundException will be caught by global handler if app not found
app = await crud.get_app(db, app_id=app_id)
if app is None:
raise HTTPException(status_code=404, detail="App not found")
return app
@router.get("/name/{name}", response_model=schemas.App)
async def get_app_by_name(name: str, db=db):
"""Get an App by Name"""
# ResourceNotFoundException will be caught by global handler if app not found
app = await crud.get_app_by_name(db, name=name)
if app is None:
raise HTTPException(status_code=404, detail="App not found")
return app
@router.post("", response_model=schemas.App)
async def create_app(app: schemas.AppCreate, db=db):
"""Create a new App"""
try:
honcho_app = await crud.create_app(db, app=app)
return honcho_app
except IntegrityError as e:
raise HTTPException(
status_code=406, detail="App with name may already exist"
) from e
except Exception as e:
print(traceback.format_exc())
raise HTTPException(status_code=400, detail="Unknown Error") from e
honcho_app = await crud.create_app(db, app=app)
return honcho_app
@router.get("/get_or_create/{name}", response_model=schemas.App)
async def get_or_create_app(name: str, db=db):
"""Get or Create an App"""
app = await crud.get_app_by_name(db=db, name=name)
if app is None:
try:
app = await crud.get_app_by_name(db=db, name=name)
return app
except ResourceNotFoundException:
# App doesn't exist, create it
app = await create_app(db=db, app=schemas.AppCreate(name=name))
return app
return app
@router.put("/{app_id}", response_model=schemas.App)
@ -63,7 +58,6 @@ async def update_app(
db=db,
):
"""Update an App"""
# ResourceNotFoundException will be caught by global handler if app not found
honcho_app = await crud.update_app(db, app_id=app_id, app=app)
if honcho_app is None:
raise HTTPException(status_code=404, detail="App not found")
return honcho_app

70
src/routers/collections.py
View File

@ -1,6 +1,6 @@
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
@ -43,10 +43,6 @@ async def get_collection_by_name(
honcho_collection = await crud.get_collection_by_name(
db, app_id=app_id, user_id=user_id, name=name
)
if honcho_collection is None:
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
)
return honcho_collection
@ -61,10 +57,6 @@ async def get_collection_by_id(
honcho_collection = await crud.get_collection_by_id(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
if honcho_collection is None:
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
)
return honcho_collection
@ -76,20 +68,11 @@ async def create_collection(
db=db,
):
"""Create a new Collection"""
# if collection.name == "honcho":
# raise HTTPException(
# status_code=406,
# detail="error invalid collection configuration - honcho is a reserved name",
# )
try:
return await crud.create_collection(
db, collection=collection, app_id=app_id, user_id=user_id
)
except ValueError:
raise HTTPException(
status_code=406,
detail="Error invalid collection configuration - name may already exist",
) from None
# ValidationException will be caught by global handler if collection is invalid
# ConflictException will be caught by global handler if collection name already exists
return await crud.create_collection(
db, collection=collection, app_id=app_id, user_id=user_id
)
@router.put("/{collection_id}", response_model=schemas.Collection)
@ -101,24 +84,15 @@ async def update_collection(
db=db,
):
"Update a Collection's name or metadata"
if collection.name is None and collection.metadata is None:
raise HTTPException(
status_code=406,
detail="error invalid collection configuration - atleast 1 field must be provided",
)
try:
honcho_collection = await crud.update_collection(
db,
collection=collection,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
)
except ValueError:
raise HTTPException(
status_code=406,
detail="Error invalid collection configuration - name may already exist",
) from None
# ResourceNotFoundException will be caught by global handler if collection not found
# ValidationException will be caught by global handler if update data is invalid
honcho_collection = await crud.update_collection(
db,
collection=collection,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
)
return honcho_collection
@ -130,12 +104,8 @@ async def delete_collection(
db=db,
):
"""Delete a Collection and its documents"""
try:
await crud.delete_collection(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
return {"message": "Collection deleted successfully"}
except ValueError:
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
) from None
# ResourceNotFoundException will be caught by global handler if collection not found
await crud.delete_collection(
db, app_id=app_id, user_id=user_id, collection_id=collection_id
)
return {"message": "Collection deleted successfully"}

79
src/routers/documents.py
View File

@ -1,15 +1,19 @@
import json
import logging
from collections.abc import Sequence
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from src import crud, schemas
from src.dependencies import db
from src.exceptions import ResourceNotFoundException, ValidationException
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections/{collection_id}/documents",
tags=["documents"],
@ -28,23 +32,19 @@ async def get_documents(
):
"""Get all of the Documents in a Collection"""
try:
return await paginate(
documents_query = await crud.get_documents(
db,
await crud.get_documents(
db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
filter=options.filter,
reverse=reverse,
),
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
filter=options.filter,
reverse=reverse,
)
except (
ValueError
): # TODO can probably remove this exception ok to return empty here
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
) from None
return await paginate(db, documents_query)
except ValueError as e:
logger.warning(f"Failed to get documents for collection {collection_id}: {str(e)}")
raise ResourceNotFoundException("Collection not found or does not belong to user") from e
@router.get(
@ -66,10 +66,6 @@ async def get_document(
collection_id=collection_id,
document_id=document_id,
)
if honcho_document is None:
raise HTTPException(
status_code=404, detail="document not found or does not belong to user"
)
return honcho_document
@ -88,7 +84,8 @@ async def query_documents(
filter = options.filter
if options.filter == {}:
filter = None
return await crud.query_documents(
documents = await crud.query_documents(
db=db,
app_id=app_id,
user_id=user_id,
@ -97,8 +94,12 @@ async def query_documents(
filter=filter,
top_k=top_k,
)
logger.info(f"Query documents successful for collection {collection_id}")
return documents
except ValueError as e:
raise HTTPException(status_code=400, detail="Error Query Documents") from e
logger.error(f"Error querying documents in collection {collection_id}: {str(e)}")
raise ValidationException("Error querying documents") from e
@router.post("", response_model=schemas.Document)
@ -111,17 +112,18 @@ async def create_document(
):
"""Embed text as a vector and create a Document"""
try:
return await crud.create_document(
document_obj = await crud.create_document(
db,
document=document,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
)
except ValueError:
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
) from None
logger.info(f"Document created successfully in collection {collection_id}")
return document_obj
except ValueError as e:
logger.warning(f"Failed to create document in collection {collection_id}: {str(e)}")
raise ResourceNotFoundException("Collection not found or does not belong to user") from e
@router.put(
@ -138,11 +140,11 @@ async def update_document(
):
"""Update the content and/or the metadata of a Document"""
if document.content is None and document.metadata is None:
raise HTTPException(
status_code=400, detail="content and metadata cannot both be None"
)
logger.warning(f"Document update attempted with empty content and metadata for document {document_id}")
raise ValidationException("Content and metadata cannot both be None")
try:
return await crud.update_document(
updated_document = await crud.update_document(
db,
document=document,
app_id=app_id,
@ -150,10 +152,11 @@ async def update_document(
collection_id=collection_id,
document_id=document_id,
)
except ValueError:
raise HTTPException(
status_code=404, detail="collection not found or does not belong to user"
) from None
logger.info(f"Document {document_id} updated successfully")
return updated_document
except ValueError as e:
logger.warning(f"Failed to update document {document_id}: {str(e)}")
raise ResourceNotFoundException("Collection or document not found") from e
@router.delete("/{document_id}")
@ -173,8 +176,8 @@ async def delete_document(
document_id=document_id,
)
if response:
logger.info(f"Document {document_id} deleted successfully")
return {"message": "Document deleted successfully"}
else:
raise HTTPException(
status_code=404, detail="document not found or does not belong to user"
)
logger.warning(f"Document {document_id} not found or could not be deleted")
raise ResourceNotFoundException("Document not found or does not belong to user")

172
src/routers/messages.py
View File

@ -1,6 +1,8 @@
from typing import Optional, List
import logging
import os
from typing import List, Optional
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
from fastapi import APIRouter, BackgroundTasks, Depends
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from sqlalchemy.sql import insert
@ -8,9 +10,12 @@ from sqlalchemy.sql import insert
from src import crud, schemas
from src.db import SessionLocal
from src.dependencies import db
from src.exceptions import ResourceNotFoundException
from src.models import QueueItem
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions/{session_id}/messages",
tags=["messages"],
@ -19,29 +24,45 @@ router = APIRouter(
async def enqueue(payload: dict | list[dict]):
"""
Add message(s) to the deriver queue for processing.
Args:
payload: Single message payload or list of message payloads
"""
async with SessionLocal() as db:
try:
if isinstance(payload, list):
if not payload: # Empty list check
logger.debug("Empty payload list, skipping enqueue")
return
print("Payload:\n", payload)
logger.debug(f"Enqueueing batch of {len(payload)} messages")
# Check session once since all messages are for same session
session = await crud.get_session(
db,
app_id=payload[0]["app_id"],
user_id=payload[0]["user_id"],
session_id=payload[0]["session_id"],
)
print("Session found:", session is not None)
if session:
print("Session metadata:", session.h_metadata)
try:
session = await crud.get_session(
db,
app_id=payload[0]["app_id"],
user_id=payload[0]["user_id"],
session_id=payload[0]["session_id"],
)
except ResourceNotFoundException:
logger.warning(
f"Session {payload[0]['session_id']} not found, skipping enqueue"
)
return
if session is None or (
logger.debug(f"Session {session.public_id} found for batch enqueue")
# Check if deriver is disabled for this session
if (
session.h_metadata.get("deriver_disabled") is not None
and session.h_metadata.get("deriver_disabled") is not False
):
print("Skipping enqueue due to session check")
logger.info(
f"Deriver is disabled for session {session.public_id}, skipping enqueue"
)
return
# Process all payloads
@ -55,32 +76,39 @@ async def enqueue(payload: dict | list[dict]):
for p in payload
]
print("Number of queue records to insert:", len(queue_records))
logger.debug(f"Inserting {len(queue_records)} queue records")
# Use insert to maintain order
stmt = insert(QueueItem).returning(QueueItem)
result = await db.execute(stmt, queue_records)
await db.execute(stmt, queue_records)
await db.commit()
print("Queue items inserted successfully")
logger.info(f"Successfully enqueued batch of {len(payload)} messages")
return
else:
# Original single insert logic
session = await crud.get_session(
db,
app_id=payload["app_id"],
user_id=payload["user_id"],
session_id=payload["session_id"],
# Single message enqueue
logger.debug(
f"Enqueueing single message for session {payload['session_id']}"
)
if session is not None:
deriver_disabled = session.h_metadata.get("deriver_disabled")
if deriver_disabled is not None and deriver_disabled is not False:
print("=====================")
print(
f"Deriver is not enabled on session {payload['session_id']}"
)
print("=====================")
return
else:
try:
session = await crud.get_session(
db,
app_id=payload["app_id"],
user_id=payload["user_id"],
session_id=payload["session_id"],
)
except ResourceNotFoundException:
logger.warning(
f"Session {payload['session_id']} not found, skipping enqueue"
)
return
# Check if deriver is disabled for this session
deriver_disabled = session.h_metadata.get("deriver_disabled")
if deriver_disabled is not None and deriver_disabled is not False:
logger.info(
f"Deriver is disabled for session {payload['session_id']}, skipping enqueue"
)
return
processed_payload = {
@ -94,12 +122,16 @@ async def enqueue(payload: dict | list[dict]):
)
await db.execute(stmt)
await db.commit()
logger.info(
f"Successfully enqueued message for session {payload['session_id']}"
)
return
except Exception as e:
print("=====================")
print("FAILURE: in enqueue")
print("=====================")
print(e)
logger.error(f"Failed to enqueue message: {str(e)}", exc_info=True)
if os.getenv("SENTRY_ENABLED", "False").lower() == "true":
import sentry_sdk
sentry_sdk.capture_exception(e)
await db.rollback()
@ -117,6 +149,8 @@ async def create_message_for_session(
honcho_message = await crud.create_message(
db, message=message, app_id=app_id, user_id=user_id, session_id=session_id
)
# Prepare message payload for background processing
payload = {
"app_id": app_id,
"user_id": user_id,
@ -126,14 +160,17 @@ async def create_message_for_session(
"content": honcho_message.content,
"metadata": honcho_message.h_metadata,
}
# Queue message for background processing
background_tasks.add_task(enqueue, payload) # type: ignore
logger.info(
f"Message {honcho_message.public_id} created and queued for processing"
)
return honcho_message
except ValueError:
print("=====================")
print("FAILURE: in create message")
print("=====================")
raise HTTPException(status_code=404, detail="Session not found") from None
except ValueError as e:
logger.error(f"Failed to create message for session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router.post("/batch", response_model=List[schemas.Message])
@ -148,7 +185,11 @@ async def create_batch_messages_for_session(
"""Bulk create messages for a session while maintaining order. Maximum 100 messages per batch."""
try:
created_messages = await crud.create_messages(
db, messages=batch.messages, app_id=app_id, user_id=user_id, session_id=session_id
db,
messages=batch.messages,
app_id=app_id,
user_id=user_id,
session_id=session_id,
)
# Create payloads for all messages
@ -167,10 +208,16 @@ async def create_batch_messages_for_session(
# Enqueue all messages in one call
background_tasks.add_task(enqueue, payloads) # type: ignore
logger.info(
f"Batch of {len(created_messages)} messages created and queued for processing"
)
return created_messages
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
except ValueError as e:
logger.error(
f"Failed to create batch messages for session {session_id}: {str(e)}"
)
raise ResourceNotFoundException("Session not found") from e
@router.post("/list", response_model=Page[schemas.Message])
@ -187,19 +234,20 @@ async def get_messages(
filter = options.filter
if options.filter == {}:
filter = None
return await paginate(
messages_query = await crud.get_messages(
db,
await crud.get_messages(
db,
app_id=app_id,
user_id=user_id,
session_id=session_id,
filter=filter,
reverse=reverse,
),
app_id=app_id,
user_id=user_id,
session_id=session_id,
filter=filter,
reverse=reverse,
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
return await paginate(db, messages_query)
except ValueError as e:
logger.warning(f"Failed to get messages for session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router.get("/{message_id}", response_model=schemas.Message)
@ -215,7 +263,8 @@ async def get_message(
db, app_id=app_id, session_id=session_id, user_id=user_id, message_id=message_id
)
if honcho_message is None:
raise HTTPException(status_code=404, detail="Message not found")
logger.warning(f"Message {message_id} not found in session {session_id}")
raise ResourceNotFoundException(f"Message with ID {message_id} not found")
return honcho_message
@ -230,7 +279,7 @@ async def update_message(
):
"""Update the metadata of a Message"""
try:
return await crud.update_message(
updated_message = await crud.update_message(
db,
message=message,
app_id=app_id,
@ -238,5 +287,8 @@ async def update_message(
session_id=session_id,
message_id=message_id,
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
logger.info(f"Message {message_id} updated successfully")
return updated_message
except ValueError as e:
logger.warning(f"Failed to update message {message_id}: {str(e)}")
raise ResourceNotFoundException("Message or session not found") from e

81
src/routers/metamessages.py
View File

@ -1,13 +1,17 @@
import logging
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from src import crud, schemas
from src.dependencies import db
from src.exceptions import ResourceNotFoundException, ValidationException
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions/{session_id}/metamessages",
tags=["metamessages"],
@ -31,15 +35,18 @@ async def create_metamessage(
):
"""Adds a message to a session"""
try:
return await crud.create_metamessage(
metamessage_obj = await crud.create_metamessage(
db,
metamessage=metamessage,
app_id=app_id,
user_id=user_id,
session_id=session_id,
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
logger.info(f"Metamessage created successfully for session {session_id}")
return metamessage_obj
except ValueError as e:
logger.warning(f"Failed to create metamessage for session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router.post("/list", response_model=Page[schemas.Metamessage])
@ -53,21 +60,21 @@ async def get_metamessages(
):
"""Get all messages for a session"""
try:
return await paginate(
metamessages_query = await crud.get_metamessages(
db,
await crud.get_metamessages(
db,
app_id=app_id,
user_id=user_id,
session_id=session_id,
message_id=options.message_id,
metamessage_type=options.metamessage_type,
filter=options.filter,
reverse=reverse,
),
app_id=app_id,
user_id=user_id,
session_id=session_id,
message_id=options.message_id,
metamessage_type=options.metamessage_type,
filter=options.filter,
reverse=reverse,
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
return await paginate(db, metamessages_query)
except ValueError as e:
logger.warning(f"Failed to get metamessages for session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router_user_level.post("/list", response_model=Page[schemas.Metamessage])
@ -80,19 +87,19 @@ async def get_metamessages_by_user(
):
"""Paginate through the user metamessages for a user"""
try:
return await paginate(
metamessages_query = await crud.get_metamessages(
db,
await crud.get_metamessages(
db,
app_id=app_id,
user_id=user_id,
metamessage_type=options.metamessage_type,
reverse=reverse,
filter=options.filter,
),
app_id=app_id,
user_id=user_id,
metamessage_type=options.metamessage_type,
reverse=reverse,
filter=options.filter,
)
except ValueError:
raise HTTPException(status_code=404, detail="User not found") from None
return await paginate(db, metamessages_query)
except ValueError as e:
logger.warning(f"Failed to get metamessages for user {user_id}: {str(e)}")
raise ResourceNotFoundException("User not found") from e
@router.get(
@ -117,7 +124,8 @@ async def get_metamessage(
metamessage_id=metamessage_id,
)
if honcho_metamessage is None:
raise HTTPException(status_code=404, detail="Session not found")
logger.warning(f"Metamessage {metamessage_id} not found for message {message_id}")
raise ResourceNotFoundException(f"Metamessage with ID {metamessage_id} not found")
return honcho_metamessage
@ -135,11 +143,11 @@ async def update_metamessage(
):
"""Update's the metadata of a metamessage"""
if metamessage.metadata is None:
raise HTTPException(
status_code=400, detail="Metamessage metadata cannot be empty"
)
logger.warning(f"Update attempted with empty metadata for metamessage {metamessage_id}")
raise ValidationException("Metamessage metadata cannot be empty")
try:
return await crud.update_metamessage(
updated_metamessage = await crud.update_metamessage(
db,
metamessage=metamessage,
app_id=app_id,
@ -147,5 +155,8 @@ async def update_metamessage(
session_id=session_id,
metamessage_id=metamessage_id,
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
logger.info(f"Metamessage {metamessage_id} updated successfully")
return updated_metamessage
except ValueError as e:
logger.warning(f"Failed to update metamessage {metamessage_id}: {str(e)}")
raise ResourceNotFoundException("Session or metamessage not found") from e

33
src/routers/sessions.py
View File

@ -1,15 +1,19 @@
import logging
from typing import Optional
from anthropic import MessageStreamManager
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from fastapi.responses import StreamingResponse
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from src import agent, crud, schemas
from src.dependencies import db
from src.exceptions import ResourceNotFoundException, ValidationException
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions",
tags=["sessions"],
@ -48,15 +52,14 @@ async def create_session(
):
"""Create a Session for a User"""
try:
value = await crud.create_session(
session_obj = await crud.create_session(
db, app_id=app_id, user_id=user_id, session=session
)
return value
logger.info(f"Session created successfully for user {user_id}")
return session_obj
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e)) from e
# except Exception as e:
# print(e)
# raise HTTPException(status_code=400, detail=str(e)) from e
logger.warning(f"Failed to create session: {str(e)}")
raise ValidationException(str(e)) from e
@router.put("/{session_id}", response_model=schemas.Session)
@ -69,11 +72,14 @@ async def update_session(
):
"""Update the metadata of a Session"""
try:
return await crud.update_session(
updated_session = await crud.update_session(
db, app_id=app_id, user_id=user_id, session_id=session_id, session=session
)
except ValueError:
raise HTTPException(status_code=404, detail="Session not found") from None
logger.info(f"Session {session_id} updated successfully")
return updated_session
except ValueError as e:
logger.warning(f"Failed to update session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router.delete("/{session_id}")
@ -88,9 +94,11 @@ async def delete_session(
await crud.delete_session(
db, app_id=app_id, user_id=user_id, session_id=session_id
)
logger.info(f"Session {session_id} deleted successfully")
return {"message": "Session deleted successfully"}
except ValueError as e:
raise HTTPException(status_code=404, detail="Session not found") from e
logger.warning(f"Failed to delete session {session_id}: {str(e)}")
raise ResourceNotFoundException("Session not found") from e
@router.get("/{session_id}", response_model=schemas.Session)
@ -105,7 +113,8 @@ async def get_session(
db, app_id=app_id, session_id=session_id, user_id=user_id
)
if honcho_session is None:
raise HTTPException(status_code=404, detail="Session not found")
logger.warning(f"Session {session_id} not found for user {user_id}")
raise ResourceNotFoundException(f"Session with ID {session_id} not found")
return honcho_session

37
src/routers/users.py
View File

@ -1,14 +1,18 @@
from typing import Optional
import logging
from fastapi import APIRouter, Depends, HTTPException
from fastapi import APIRouter, Depends
from fastapi_pagination import Page
from fastapi_pagination.ext.sqlalchemy import paginate
from sqlalchemy.exc import IntegrityError
from src import crud, schemas
from src.dependencies import db
from src.exceptions import (
ResourceNotFoundException,
)
from src.security import auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users",
tags=["users"],
@ -23,12 +27,8 @@ async def create_user(
db=db,
):
"""Create a new User"""
try:
return await crud.create_user(db, app_id=app_id, user=user)
except IntegrityError as e:
raise HTTPException(
status_code=406, detail="User with name may already exist"
) from e
user_obj = await crud.create_user(db, app_id=app_id, user=user)
return user_obj
@router.post("/list", response_model=Page[schemas.User])
@ -53,8 +53,6 @@ async def get_user_by_name(
):
"""Get a User by name"""
user = await crud.get_user_by_name(db, app_id=app_id, name=name)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
@ -66,20 +64,21 @@ async def get_user(
):
"""Get a User by ID"""
user = await crud.get_user(db, app_id=app_id, user_id=user_id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.get("/get_or_create/{name}", response_model=schemas.User)
async def get_or_create_user(app_id: str, name: str, db=db):
"""Get a User or create a new one by the input name"""
user = await crud.get_user_by_name(db, app_id=app_id, name=name)
if user is None:
try:
user = await crud.get_user_by_name(db, app_id=app_id, name=name)
return user
except ResourceNotFoundException:
# User doesn't exist, create it
user = await create_user(
db=db, app_id=app_id, user=schemas.UserCreate(name=name)
)
return user
return user
@router.put("/{user_id}", response_model=schemas.User)
@ -90,7 +89,5 @@ async def update_user(
db=db,
):
"""Update a User's name and/or metadata"""
try:
return await crud.update_user(db, app_id=app_id, user_id=user_id, user=user)
except ValueError as e:
raise HTTPException(status_code=406, detail=str(e)) from e
updated_user = await crud.update_user(db, app_id=app_id, user_id=user_id, user=user)
return updated_user

10
src/security.py
View File

@ -1,9 +1,14 @@
import logging
import os
from typing import Annotated
from fastapi import Depends, HTTPException
from fastapi import Depends
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from .exceptions import AuthenticationException
logger = logging.getLogger(__name__)
USE_AUTH_SERVICE = os.getenv("USE_AUTH_SERVICE", "False").lower() == "true"
SECRET_KEY = os.getenv("SECRET_KEY", "test")
@ -18,5 +23,6 @@ async def auth(
if not USE_AUTH_SERVICE:
return True
if not credentials or credentials.credentials != SECRET_KEY:
raise HTTPException(status_code=401, detail="Invalid access token")
logger.warning("Invalid access token attempt")
raise AuthenticationException("Invalid access token")
return {"message": "OK"}

11
tests/conftest.py
View File

@ -5,6 +5,8 @@ from nanoid import generate as generate_nanoid
import pytest
import pytest_asyncio
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine, AsyncSession
@ -15,6 +17,7 @@ from sqlalchemy_utils import create_database, database_exists, drop_database
from src import models
from src.db import Base
from src.dependencies import get_db
from src.exceptions import HonchoException
from src.main import app
logging.basicConfig(
@ -114,6 +117,14 @@ async def db_session(db_engine):
def client(db_session):
"""Create a FastAPI TestClient for the scope of a single test function"""
# Register exception handlers for tests
@app.exception_handler(HonchoException)
async def test_exception_handler(request: Request, exc: HonchoException):
return JSONResponse(
status_code=exc.status_code,
content={"detail": exc.detail},
)
async def override_get_db():
yield db_session

27
tests/routes/test_apps.py
View File

@ -33,8 +33,12 @@ def test_create_app_no_metadata(client):
def test_get_or_create_app(client):
name = str(generate_nanoid())
# Should return a ResourceNotFoundException with 404 status
response = client.get(f"/v1/apps/name/{name}")
assert response.status_code == 404
assert "detail" in response.json()
# This should create the app
response = client.get(f"/v1/apps/get_or_create/{name}")
assert response.status_code == 200
data = response.json()
@ -44,16 +48,24 @@ def test_get_or_create_app(client):
def test_get_or_create_existing_app(client):
name = str(generate_nanoid())
# App doesn't exist yet
response = client.get(f"/v1/apps/name/{name}")
assert response.status_code == 404
# Create the app
response = client.post(
"/v1/apps", json={"name": name, "metadata": {"key": "value"}}
)
assert response.status_code == 200
app1 = response.json()
# Now get_or_create should find the existing app
response = client.get(f"/v1/apps/get_or_create/{name}")
assert response.status_code == 200
app2 = response.json()
# Both should be the same app
assert app1["name"] == app2["name"]
assert app1["id"] == app2["id"]
assert app1["metadata"] == app2["metadata"]
@ -88,3 +100,18 @@ def test_update_app(client, sample_data):
data = response.json()
assert data["name"] == new_name
assert data["metadata"] == {"new_key": "new_value"}
def test_create_duplicate_app_name(client):
# Create an app
name = str(generate_nanoid())
response = client.post("/v1/apps", json={"name": name})
assert response.status_code == 200
# Try to create another app with the same name
response = client.post("/v1/apps", json={"name": name})
# Should get a ConflictException with 409 status
assert response.status_code == 409
data = response.json()
assert "detail" in data

30
tests/routes/test_documents.py
View File

@ -1,6 +1,36 @@
from nanoid import generate as generate_nanoid
def test_update_document_validation_error(client, sample_data):
test_app, test_user = sample_data
# Create a collection
response = client.post(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections",
json={"name": "test_collection", "metadata": {}},
)
assert response.status_code == 200
collection = response.json()
# Create a document
response = client.post(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{collection['id']}/documents",
json={"content": "test_text", "metadata": {}},
)
assert response.status_code == 200
document = response.json()
# Try to update the document with empty content and metadata
response = client.put(
f"/v1/apps/{test_app.public_id}/users/{test_user.public_id}/collections/{collection['id']}/documents/{document['id']}",
json={"content": None, "metadata": None},
)
# Should get a ValidationException with 422 status
assert response.status_code == 422
data = response.json()
assert "detail" in data
def test_create_document(client, sample_data):
test_app, test_user = sample_data
# Create a collection

6
uv.lock
View File

@ -1262,15 +1262,15 @@ wheels = [
[[package]]
name = "sentry-sdk"
version = "2.19.2"
version = "2.22.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "certifi" },
{ name = "urllib3" },
]
sdist = { url = "https://files.pythonhosted.org/packages/36/4a/eccdcb8c2649d53440ae1902447b86e2e2ad1bc84207c80af9696fa07614/sentry_sdk-2.19.2.tar.gz", hash = "sha256:467df6e126ba242d39952375dd816fbee0f217d119bf454a8ce74cf1e7909e8d", size = 299047 }
sdist = { url = "https://files.pythonhosted.org/packages/81/b6/662988ecd2345bf6c3a5c306a9a3590852742eff91d0a78a143398b816f3/sentry_sdk-2.22.0.tar.gz", hash = "sha256:b4bf43bb38f547c84b2eadcefbe389b36ef75f3f38253d7a74d6b928c07ae944", size = 303539 }
wheels = [
{ url = "https://files.pythonhosted.org/packages/31/4d/74597bb6bcc23abc774b8901277652c61331a9d4d0a8d1bdb20679b9bbcb/sentry_sdk-2.19.2-py2.py3-none-any.whl", hash = "sha256:ebdc08228b4d131128e568d696c210d846e5b9d70aa0327dec6b1272d9d40b84", size = 322942 },
{ url = "https://files.pythonhosted.org/packages/12/7f/0e4459173e9671ba5f75a48dda2442bcc48a12c79e54e5789381c8c6a9bc/sentry_sdk-2.22.0-py2.py3-none-any.whl", hash = "sha256:3d791d631a6c97aad4da7074081a57073126c69487560c6f8bffcf586461de66", size = 325815 },
]
[package.optional-dependencies]