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

update routers

This commit is contained in:
Rajat Ahuja 2025-06-10 13:22:30 -04:00
parent 257c7f5542
commit 4f59d63d41
10 changed files with 397 additions and 927 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/main.py
View File

@ -15,14 +15,13 @@ from sentry_sdk.integrations.starlette import StarletteIntegration
from src.db import engine, request_context
from src.exceptions import HonchoException
from src.routers import (
apps,
collections,
documents,
keys,
messages,
metamessages,
peers,
sessions,
users,
workspaces,
)
from src.security import create_admin_jwt
@ -138,11 +137,10 @@ router = APIRouter(prefix="/apps/{app_id}/users/{user_id}")
add_pagination(app)
app.include_router(apps.router, prefix="/v1")
app.include_router(users.router, prefix="/v1")
app.include_router(workspaces.router, prefix="/v1")
app.include_router(peers.router, prefix="/v1")
app.include_router(sessions.router, prefix="/v1")
app.include_router(messages.router, prefix="/v1")
app.include_router(metamessages.router, prefix="/v1")
app.include_router(collections.router, prefix="/v1")
app.include_router(documents.router, prefix="/v1")
app.include_router(keys.router, prefix="/v1")

138
src/routers/apps.py
View File

@ -1,138 +0,0 @@
import logging
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 AuthenticationException, ResourceNotFoundException
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps",
tags=["apps"],
)
@router.get("", response_model=schemas.App)
async def get_app(
app_id: Optional[str] = Query(
None, description="App ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get an App by ID.
If app_id is provided as a query parameter, it uses that (must match JWT app_id).
Otherwise, it uses the app_id from the JWT token.
"""
# If app_id provided in query, check if it matches jwt or user is admin
if app_id:
if not jwt_params.ad and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
target_app_id = app_id
else:
# Use app_id from JWT
if not jwt_params.ap:
raise AuthenticationException("App ID not found in query parameter or JWT")
target_app_id = jwt_params.ap
return await crud.get_app(db, app_id=target_app_id)
@router.post(
"/list",
response_model=Page[schemas.App],
dependencies=[Depends(require_auth(admin=True))],
)
async def get_all_apps(
options: Optional[schemas.AppGet] = Body(
None, description="Filtering and pagination options for the apps list"
),
reverse: Optional[bool] = Query(
False, description="Whether to reverse the order of results"
),
db=db,
):
"""Get all Apps"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}:
filter_param = None
return await paginate(
db,
await crud.get_all_apps(
reverse=reverse,
filter=filter_param,
),
)
@router.get(
"/name/{name}",
response_model=schemas.App,
dependencies=[Depends(require_auth(admin=True))],
)
async def get_app_by_name(
name: str = Path(..., description="Name of the app to retrieve"),
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)
return app
@router.post(
"", response_model=schemas.App, dependencies=[Depends(require_auth(admin=True))]
)
async def create_app(
app: schemas.AppCreate = Body(..., description="App creation parameters"),
db=db,
):
"""Create a new App"""
honcho_app = await crud.create_app(db, app=app)
return honcho_app
@router.get(
"/get_or_create/{name}",
response_model=schemas.App,
dependencies=[Depends(require_auth(admin=True))],
)
async def get_or_create_app(
name: str = Path(..., description="Name of the app to get or create"),
db=db,
):
"""Get or Create an App"""
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
@router.put(
"/{app_id}",
response_model=schemas.App,
dependencies=[Depends(require_auth(app_id="app_id"))],
)
async def update_app(
app_id: str = Path(..., description="ID of the app to update"),
app: schemas.AppUpdate = Body(..., description="Updated app parameters"),
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)
return honcho_app

190
src/routers/collections.py
View File

@ -1,190 +0,0 @@
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 AuthenticationException
from src.security import JWTParams, require_auth
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections",
tags=["collections"],
)
@router.get(
"",
response_model=schemas.Collection,
)
async def get_collection(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: Optional[str] = Query(
None, description="Collection ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a specific collection for a user.
If collection_id is provided as a query parameter, it uses that (must match JWT collection_id).
Otherwise, it uses the collection_id from the JWT token.
"""
# Verify JWT has access to the requested resource
if not jwt_params.ad:
if jwt_params.ap is not None and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
if jwt_params.us is not None and jwt_params.us != user_id:
raise AuthenticationException("Unauthorized access to resource")
# If collection_id provided in query, check if it matches jwt or user is admin
if collection_id:
if (
not jwt_params.ad
and jwt_params.co is not None
and jwt_params.co != collection_id
):
raise AuthenticationException("Unauthorized access to resource")
target_collection_id = collection_id
else:
# Use collection_id from JWT
if not jwt_params.co:
raise AuthenticationException(
"Collection ID not found in query parameter or JWT"
)
target_collection_id = jwt_params.co
# Let crud function handle the ResourceNotFoundException
return await crud.get_collection_by_id(
db, app_id=app_id, collection_id=target_collection_id, user_id=user_id
)
@router.post(
"/list",
response_model=Page[schemas.Collection],
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def get_collections(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
options: Optional[schemas.CollectionGet] = Body(
None, description="Filtering options for the collections list"
),
reverse: Optional[bool] = Query(
False, description="Whether to reverse the order of results"
),
db=db,
):
"""Get All Collections for a User"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}: # Explicitly check for empty dict
filter_param = None
return await paginate(
db,
await crud.get_collections(
app_id=app_id, user_id=user_id, filter=filter_param, reverse=reverse
),
)
@router.get(
"/name/{name}",
response_model=schemas.Collection,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def get_collection_by_name(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
name: str = Path(..., description="Name of the collection to retrieve"),
db=db,
) -> schemas.Collection:
"""Get a Collection by Name"""
honcho_collection = await crud.get_collection_by_name(
db, app_id=app_id, user_id=user_id, name=name
)
return honcho_collection
@router.post(
"",
response_model=schemas.Collection,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def create_collection(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection: schemas.CollectionCreate = Body(
..., description="Collection creation parameters"
),
db=db,
):
"""Create a new Collection"""
# 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,
dependencies=[
Depends(
require_auth(
app_id="app_id", user_id="user_id", collection_id="collection_id"
)
)
],
)
async def update_collection(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection to update"),
collection: schemas.CollectionUpdate = Body(
..., description="Updated collection parameters"
),
db=db,
):
"Update a Collection's name or metadata"
# 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
@router.delete(
"/{collection_id}",
dependencies=[
Depends(
require_auth(
app_id="app_id", user_id="user_id", collection_id="collection_id"
)
)
],
)
async def delete_collection(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection to delete"),
db=db,
):
"""Delete a Collection and its documents"""
# 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"}

212
src/routers/documents.py
View File

@ -1,212 +0,0 @@
import logging
from collections.abc import Sequence
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/collections/{collection_id}/documents",
tags=["documents"],
dependencies=[
Depends(
require_auth(
app_id="app_id", user_id="user_id", collection_id="collection_id"
)
)
],
)
@router.post("/list", response_model=Page[schemas.Document])
async def get_documents(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
options: Optional[schemas.DocumentGet] = Body(
None, description="Filtering options for the documents list"
),
reverse: Optional[bool] = Query(
False, description="Whether to reverse the order of results"
),
db=db,
):
"""Get all of the Documents in a Collection"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}: # Explicitly check for empty dict
filter_param = None
try:
documents_query = await crud.get_documents(
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
filter=filter_param,
reverse=reverse,
)
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("/{document_id}", response_model=schemas.Document)
async def get_document(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
document_id: str = Path(..., description="ID of the document to retrieve"),
db=db,
):
"""Get a document by ID"""
honcho_document = await crud.get_document(
db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
document_id=document_id,
)
return honcho_document
@router.post("/query", response_model=Sequence[schemas.Document])
async def query_documents(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
options: schemas.DocumentQuery = Body(
..., description="Query parameters for document search"
),
db=db,
):
"""Cosine Similarity Search for Documents"""
try:
top_k = options.top_k
filter = options.filter
if options.filter == {}:
filter = None
documents = await crud.query_documents(
db=db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
query=options.query,
filter=filter,
top_k=top_k,
)
logger.info(f"Query documents successful for collection {collection_id}")
return documents
except ValueError as 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)
async def create_document(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
document: schemas.DocumentCreate = Body(
..., description="Document creation parameters"
),
db=db,
):
"""Embed text as a vector and create a Document"""
try:
document_obj = await crud.create_document(
db,
document=document,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
)
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(
"/{document_id}",
response_model=schemas.Document,
)
async def update_document(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
document_id: str = Path(..., description="ID of the document to update"),
document: schemas.DocumentUpdate = Body(
..., description="Updated document parameters"
),
db=db,
):
"""Update the content and/or the metadata of a Document"""
if document.content is None and document.metadata is 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:
updated_document = await crud.update_document(
db,
document=document,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
document_id=document_id,
)
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}")
async def delete_document(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
collection_id: str = Path(..., description="ID of the collection"),
document_id: str = Path(..., description="ID of the document to delete"),
db=db,
):
"""Delete a Document by ID"""
response = await crud.delete_document(
db,
app_id=app_id,
user_id=user_id,
collection_id=collection_id,
document_id=document_id,
)
if response:
logger.info(f"Document {document_id} deleted successfully")
return {"message": "Document deleted successfully"}
else:
logger.warning(f"Document {document_id} not found or could not be deleted")
raise ResourceNotFoundException("Document not found or does not belong to user")

4
src/routers/messages.py
View File

@ -16,11 +16,11 @@ from src.security import require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions/{session_id}/messages",
prefix="/workspaces/{workspace_id}/sessions/{session_id}/messages",
tags=["messages"],
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
require_auth(app_id="workspace_id", session_id="session_id")
)
],
)

154
src/routers/metamessages.py
View File

@ -1,154 +0,0 @@
import logging
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/metamessages",
tags=["metamessages"],
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
@router.post("", response_model=schemas.Metamessage)
async def create_metamessage(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
metamessage: schemas.MetamessageCreate = Body(
..., description="Metamessage creation parameters"
),
db=db,
):
"""
Create a new metamessage associated with a user.
Optionally link to a session and message by providing those IDs in the request body.
"""
try:
metamessage_obj = await crud.create_metamessage(
db,
user_id=user_id,
metamessage=metamessage,
app_id=app_id,
)
logger.info(f"Metamessage created successfully for user {user_id}")
return metamessage_obj
except (ResourceNotFoundException, ValidationException) as e:
logger.warning(f"Failed to create metamessage: {str(e)}")
raise
@router.post("/list", response_model=Page[schemas.Metamessage])
async def get_metamessages(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
options: Optional[schemas.MetamessageGet] = Body(
None, description="Filtering options for the metamessages list"
),
reverse: Optional[bool] = Query(
False, description="Whether to reverse the order of results"
),
db=db,
):
"""
Get metamessages with flexible filtering.
- Filter by user only: No additional parameters needed
- Filter by session: Provide session_id
- Filter by message: Provide message_id (and session_id)
- Filter by type: Provide label
- Filter by metadata: Provide filter object
"""
session_id_param = None
message_id_param = None
label_param = None
filter_param = None
if options:
if hasattr(options, "session_id") and options.session_id:
session_id_param = options.session_id
if hasattr(options, "message_id") and options.message_id:
message_id_param = options.message_id
if hasattr(options, "label") and options.label:
label_param = options.label
if hasattr(options, "filter") and options.filter:
filter_param = options.filter
if filter_param == {}: # Explicitly check for empty dict
filter_param = None
try:
metamessages_query = await crud.get_metamessages(
app_id=app_id,
user_id=user_id,
session_id=session_id_param,
message_id=message_id_param,
label=label_param,
filter=filter_param,
reverse=reverse,
)
return await paginate(db, metamessages_query)
except (ResourceNotFoundException, ValidationException) as e:
logger.warning(f"Failed to get metamessages: {str(e)}")
raise
@router.get(
"/{metamessage_id}",
response_model=schemas.Metamessage,
)
async def get_metamessage(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
metamessage_id: str = Path(..., description="ID of the metamessage to retrieve"),
db=db,
):
"""Get a specific Metamessage by ID"""
honcho_metamessage = await crud.get_metamessage(
db,
app_id=app_id,
user_id=user_id,
metamessage_id=metamessage_id,
)
if honcho_metamessage is None:
logger.warning(f"Metamessage {metamessage_id} not found")
raise ResourceNotFoundException(
f"Metamessage with ID {metamessage_id} not found"
)
return honcho_metamessage
@router.put(
"/{metamessage_id}",
response_model=schemas.Metamessage,
)
async def update_metamessage(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
metamessage_id: str = Path(..., description="ID of the metamessage to update"),
metamessage: schemas.MetamessageUpdate = Body(
..., description="Updated metamessage parameters"
),
db=db,
):
"""Update a metamessage's metadata, type, or relationships"""
try:
updated_metamessage = await crud.update_metamessage(
db,
metamessage=metamessage,
app_id=app_id,
user_id=user_id,
metamessage_id=metamessage_id,
)
logger.info(f"Metamessage {metamessage_id} updated successfully")
return updated_metamessage
except (ResourceNotFoundException, ValidationException) as e:
logger.warning(f"Failed to update metamessage {metamessage_id}: {str(e)}")
raise

200
src/routers/peers.py Normal file
View File

@ -0,0 +1,200 @@
import logging
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 (
AuthenticationException,
ResourceNotFoundException,
)
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/workspaces/{workspace_id}/peers",
tags=["peers"],
)
@router.post(
"",
response_model=schemas.Peer,
dependencies=[Depends(require_auth(app_id="workspace_id"))],
)
async def create_peer(
workspace_id: str = Path(..., description="ID of the workspace"),
peer: schemas.PeerCreate = Body(..., description="Peer creation parameters"),
db=db,
):
"""Create a new Peer"""
peer_obj = await crud.create_peer(db, workspace_id=workspace_id, peer=peer)
return peer_obj
@router.post(
"/list",
response_model=Page[schemas.Peer],
dependencies=[Depends(require_auth(app_id="workspace_id"))],
)
async def get_peers(
workspace_id: str = Path(..., description="ID of the workspace"),
options: Optional[schemas.PeerGet] = Body(
None, description="Filtering options for the peers list"
),
reverse: bool = Query(False, description="Whether to reverse the order of results"),
db=db,
):
"""Get All Peers for a Workspace"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}:
filter_param = None
return await paginate(
db,
await crud.get_peers(workspace_id=workspace_id, reverse=reverse, filter=filter_param),
)
@router.get(
"",
response_model=schemas.Peer,
)
async def get_peer(
workspace_id: str = Path(..., description="ID of the workspace"),
peer_id: Optional[str] = Query(
None, description="Peer ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a Peer by ID
If peer_id is provided as a query parameter, it uses that (must match JWT workspace_id).
Otherwise, it uses the peer_id from the JWT token.
"""
# validate workspace query param
if not jwt_params.ad and jwt_params.ap is not None and jwt_params.ap != workspace_id:
raise AuthenticationException("Unauthorized access to resource")
if peer_id:
if not jwt_params.ad and jwt_params.us is not None and jwt_params.us != peer_id:
raise AuthenticationException("Unauthorized access to resource")
target_peer_id = peer_id
else:
# Use peer_id from JWT
if not jwt_params.us:
raise AuthenticationException("Peer ID not found in query parameter or JWT")
target_peer_id = jwt_params.us
peer = await crud.get_peer(db, workspace_id=workspace_id, peer_id=target_peer_id)
return peer
@router.get(
"/name/{name}",
response_model=schemas.Peer,
dependencies=[
Depends(
require_auth(
app_id="workspace_id",
)
)
],
)
async def get_peer_by_name(
workspace_id: str = Path(..., description="ID of the workspace"),
name: str = Path(..., description="Name of the peer to retrieve"),
db=db,
):
"""Get a Peer by name"""
peer = await crud.get_peer_by_name(db, workspace_id=workspace_id, name=name)
return peer
@router.get(
"/get_or_create/{name}",
response_model=schemas.Peer,
dependencies=[
Depends(
require_auth(
app_id="workspace_id",
)
)
],
)
async def get_or_create_peer(
workspace_id: str = Path(..., description="ID of the workspace"),
name: str = Path(..., description="Name of the peer to get or create"),
db=db,
):
"""Get a Peer or create a new one by the input name"""
try:
peer = await crud.get_peer_by_name(db, workspace_id=workspace_id, name=name)
return peer
except ResourceNotFoundException:
# Peer doesn't exist, create it
peer = await create_peer(
db=db, workspace_id=workspace_id, peer=schemas.PeerCreate(name=name)
)
return peer
@router.put(
"/{peer_id}",
response_model=schemas.Peer,
dependencies=[Depends(require_auth(app_id="workspace_id", user_id="peer_id"))],
)
async def update_peer(
workspace_id: str = Path(..., description="ID of the workspace"),
peer_id: str = Path(..., description="ID of the peer to update"),
peer: schemas.PeerUpdate = Body(..., description="Updated peer parameters"),
db=db,
):
"""Update a Peer's name and/or metadata"""
updated_peer = await crud.update_peer(db, workspace_id=workspace_id, peer_id=peer_id, peer=peer)
return updated_peer
@router.post(
"/{peer_id}/sessions",
response_model=Page[schemas.Session],
dependencies=[Depends(require_auth(app_id="workspace_id", user_id="peer_id"))],
)
async def get_peer_sessions(
workspace_id: str = Path(..., description="ID of the workspace"),
peer_id: str = Path(..., description="ID of the peer"),
options: Optional[schemas.SessionGet] = Body(
None, description="Filtering options for the sessions list"
),
reverse: bool = Query(False, description="Whether to reverse the order of results"),
db=db,
):
"""Get All Sessions for a Peer"""
filter_param = None
is_active = False # Default from schemas
if options:
if hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}:
filter_param = None
if hasattr(options, "is_active"):
is_active = options.is_active
return await paginate(
db,
await crud.get_peer_sessions(
workspace_id=workspace_id,
peer_id=peer_id,
reverse=reverse,
is_active=is_active,
filter=filter_param,
),
)

116
src/routers/sessions.py
View File

@ -20,67 +20,53 @@ from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users/{user_id}/sessions",
prefix="/workspaces/{workspace_id}/sessions",
tags=["sessions"],
)
@router.get(
"",
"/{session_id}",
response_model=schemas.Session,
)
async def get_session(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
session_id: Optional[str] = Query(
None, description="Session ID to retrieve. If not provided, uses JWT token"
workspace_id: str = Path(..., description="ID of the workspace"),
session_id: str = Path(..., description="ID of the session"),
peer_id: Optional[str] = Query(
None, description="Peer ID to verify access. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a specific session for a user.
Get a specific session in a workspace.
If session_id is provided as a query parameter, it uses that (must match JWT session_id).
Otherwise, it uses the session_id from the JWT token.
If peer_id is provided as a query parameter, it verifies the peer is in the session.
Otherwise, it uses the peer_id from the JWT token for verification.
"""
# Verify JWT has access to the requested resource
if not jwt_params.ad:
if jwt_params.ap is not None and jwt_params.ap != app_id:
if jwt_params.ap is not None and jwt_params.ap != workspace_id:
raise AuthenticationException("Unauthorized access to resource")
if jwt_params.us is not None and jwt_params.us != user_id:
if peer_id and jwt_params.us is not None and jwt_params.us != peer_id:
raise AuthenticationException("Unauthorized access to resource")
# If session_id provided in query, check if it matches jwt or user is admin
if session_id:
if (
not jwt_params.ad
and jwt_params.se is not None
and jwt_params.se != session_id
):
raise AuthenticationException("Unauthorized access to resource")
target_session_id = session_id
else:
# Use session_id from JWT
if not jwt_params.se:
raise AuthenticationException(
"Session ID not found in query parameter or JWT"
)
target_session_id = jwt_params.se
# Use peer_id from JWT if not provided in query
target_peer_id = peer_id or jwt_params.us
# Let crud function handle the ResourceNotFoundException
return await crud.get_session(
db, app_id=app_id, session_id=target_session_id, user_id=user_id
db, workspace_id=workspace_id, session_id=session_id, peer_id=target_peer_id
)
@router.post(
"/list",
response_model=Page[schemas.Session],
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
dependencies=[Depends(require_auth(app_id="workspace_id"))],
)
async def get_sessions(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
options: Optional[schemas.SessionGet] = Body(
None, description="Filtering and pagination options for the sessions list"
),
@ -89,9 +75,9 @@ async def get_sessions(
),
db=db,
):
"""Get All Sessions for a User"""
"""Get All Sessions in a Workspace"""
filter_param = None
is_active_param = False # Default to None, meaning no filter on is_active
is_active_param = False # Default from schema
if options:
if hasattr(options, 'filter') and options.filter:
@ -104,8 +90,7 @@ async def get_sessions(
return await paginate(
db,
await crud.get_sessions(
app_id=app_id,
user_id=user_id,
workspace_id=workspace_id,
reverse=reverse,
is_active=is_active_param,
filter=filter_param,
@ -116,22 +101,24 @@ async def get_sessions(
@router.post(
"",
response_model=schemas.Session,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
dependencies=[Depends(require_auth(app_id="workspace_id"))],
)
async def create_session(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
session: schemas.SessionCreate = Body(
..., description="Session creation parameters"
),
peer_ids: Optional[list[str]] = Body(
None, description="List of peer IDs to add to the session"
),
db=db,
):
"""Create a Session for a User"""
"""Create a Session in a Workspace"""
try:
session_obj = await crud.create_session(
db, app_id=app_id, user_id=user_id, session=session
db, workspace_id=workspace_id, session=session, peer_ids=peer_ids
)
logger.info(f"Session created successfully for user {user_id}")
logger.info(f"Session created successfully in workspace {workspace_id}")
return session_obj
except ValueError as e:
logger.warning(f"Failed to create session: {str(e)}")
@ -143,23 +130,25 @@ async def create_session(
response_model=schemas.Session,
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
require_auth(app_id="workspace_id", session_id="session_id")
)
],
)
async def update_session(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
session_id: str = Path(..., description="ID of the session to update"),
session: schemas.SessionUpdate = Body(
..., description="Updated session parameters"
),
peer_id: Optional[str] = Query(
None, description="Peer ID to verify access"
),
db=db,
):
"""Update the metadata of a Session"""
try:
updated_session = await crud.update_session(
db, app_id=app_id, user_id=user_id, session_id=session_id, session=session
db, workspace_id=workspace_id, session_id=session_id, session=session, peer_id=peer_id
)
logger.info(f"Session {session_id} updated successfully")
return updated_session
@ -172,20 +161,19 @@ async def update_session(
"/{session_id}",
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
require_auth(app_id="workspace_id", session_id="session_id")
)
],
)
async def delete_session(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
session_id: str = Path(..., description="ID of the session to delete"),
db=db,
):
"""Delete a session by marking it as inactive"""
try:
await crud.delete_session(
db, app_id=app_id, user_id=user_id, session_id=session_id
db, workspace_id=workspace_id, session_id=session_id
)
logger.info(f"Session {session_id} deleted successfully")
return {"message": "Session deleted successfully"}
@ -194,6 +182,8 @@ async def delete_session(
raise ResourceNotFoundException("Session not found") from e
# TODO: Update chat endpoint to work with new workspace/peer paradigm
# This endpoint needs significant rework for multi-peer sessions
@router.post(
"/{session_id}/chat",
response_model=schemas.DialecticResponse,
@ -205,24 +195,25 @@ async def delete_session(
},
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
require_auth(app_id="workspace_id", session_id="session_id")
)
],
)
async def chat(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
session_id: str = Path(..., description="ID of the session"),
peer_id: str = Query(..., description="ID of the peer making the request"),
options: schemas.DialecticOptions = Body(
..., description="Dialectic Endpoint Parameters"
),
):
"""Chat with the Dialectic API"""
# TODO: Update agent.chat to work with workspace_id/peer_id instead of app_id/user_id
if not options.stream:
return await agent.chat(
app_id=app_id,
user_id=user_id,
app_id=workspace_id, # Temporary mapping
user_id=peer_id, # Temporary mapping
session_id=session_id,
queries=options.queries,
)
@ -231,8 +222,8 @@ async def chat(
async def parse_stream():
try:
stream = await agent.chat(
app_id=app_id,
user_id=user_id,
app_id=workspace_id, # Temporary mapping
user_id=peer_id, # Temporary mapping
session_id=session_id,
queries=options.queries,
stream=True,
@ -250,34 +241,33 @@ async def chat(
)
# TODO: Implement clone_session endpoint for new workspace/peer paradigm
# Need to update clone_session CRUD method first
@router.get(
"/{session_id}/clone",
response_model=schemas.Session,
dependencies=[
Depends(
require_auth(app_id="app_id", user_id="user_id", session_id="session_id")
require_auth(app_id="workspace_id", session_id="session_id")
)
],
)
async def clone_session(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user"),
workspace_id: str = Path(..., description="ID of the workspace"),
session_id: str = Path(..., description="ID of the session to clone"),
db=db,
message_id: Optional[str] = Query(
None, description="Message ID to cut off the clone at"
),
deep_copy: bool = Query(False, description="Whether to deep copy metamessages"),
):
"""Clone a session, optionally up to a specific message"""
try:
# TODO: Update crud.clone_session to work with new paradigm
cloned_session = await crud.clone_session(
db,
app_id=app_id,
user_id=user_id,
workspace_id=workspace_id,
original_session_id=session_id,
cutoff_message_id=message_id,
deep_copy=deep_copy,
)
logger.info(f"Session {session_id} cloned successfully")
return cloned_session

162
src/routers/users.py
View File

@ -1,162 +0,0 @@
import logging
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 (
AuthenticationException,
ResourceNotFoundException,
)
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/apps/{app_id}/users",
tags=["users"],
)
@router.post(
"",
response_model=schemas.User,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def create_user(
app_id: str = Path(..., description="ID of the app"),
user: schemas.UserCreate = Body(..., description="User creation parameters"),
db=db,
):
"""Create a new User"""
user_obj = await crud.create_user(db, app_id=app_id, user=user)
return user_obj
@router.post(
"/list",
response_model=Page[schemas.User],
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def get_users(
app_id: str = Path(..., description="ID of the app"),
options: Optional[schemas.UserGet] = Body(
None, description="Filtering options for the users list"
),
reverse: bool = Query(False, description="Whether to reverse the order of results"),
db=db,
):
"""Get All Users for an App"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}:
filter_param = None
return await paginate(
db,
await crud.get_users(app_id=app_id, reverse=reverse, filter=filter_param),
)
@router.get(
"",
response_model=schemas.User,
)
async def get_user(
app_id: str = Path(..., description="ID of the app"),
user_id: Optional[str] = Query(
None, description="User ID to retrieve. If not provided, users JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a User by ID
If user_id is provided as a query parameter, it uses that (must match JWT app_id).
Otherwise, it uses the user_id from the JWT token.
"""
# validate app query param
if not jwt_params.ad and jwt_params.ap is not None and jwt_params.ap != app_id:
raise AuthenticationException("Unauthorized access to resource")
if user_id:
if not jwt_params.ad and jwt_params.us is not None and jwt_params.us != user_id:
raise AuthenticationException("Unauthorized access to resource")
target_user_id = user_id
else:
# Use user_id from JWT
if not jwt_params.us:
raise AuthenticationException("User ID not found in query parameter or JWT")
target_user_id = jwt_params.us
user = await crud.get_user(db, app_id=app_id, user_id=target_user_id)
return user
@router.get(
"/name/{name}",
response_model=schemas.User,
dependencies=[
Depends(
require_auth(
app_id="app_id",
)
)
],
)
async def get_user_by_name(
app_id: str = Path(..., description="ID of the app"),
name: str = Path(..., description="Name of the user to retrieve"),
db=db,
):
"""Get a User by name"""
user = await crud.get_user_by_name(db, app_id=app_id, name=name)
return user
@router.get(
"/get_or_create/{name}",
response_model=schemas.User,
dependencies=[
Depends(
require_auth(
app_id="app_id",
)
)
],
)
async def get_or_create_user(
app_id: str = Path(..., description="ID of the app"),
name: str = Path(..., description="Name of the user to get or create"),
db=db,
):
"""Get a User or create a new one by the input name"""
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
@router.put(
"/{user_id}",
response_model=schemas.User,
dependencies=[Depends(require_auth(app_id="app_id", user_id="user_id"))],
)
async def update_user(
app_id: str = Path(..., description="ID of the app"),
user_id: str = Path(..., description="ID of the user to update"),
user: schemas.UserUpdate = Body(..., description="Updated user parameters"),
db=db,
):
"""Update a User's name and/or metadata"""
updated_user = await crud.update_user(db, app_id=app_id, user_id=user_id, user=user)
return updated_user

138
src/routers/workspaces.py Normal file
View File

@ -0,0 +1,138 @@
import logging
from typing import Optional
from fastapi import APIRouter, Body, Depends, Path, Query
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 AuthenticationException, ResourceNotFoundException
from src.security import JWTParams, require_auth
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/workspaces",
tags=["workspaces"],
)
@router.get("", response_model=schemas.Workspace)
async def get_workspace(
workspace_id: Optional[str] = Query(
None, description="Workspace ID to retrieve. If not provided, uses JWT token"
),
jwt_params: JWTParams = Depends(require_auth()),
db=db,
):
"""
Get a Workspace by ID.
If workspace_id is provided as a query parameter, it uses that (must match JWT workspace_id).
Otherwise, it uses the workspace_id from the JWT token.
"""
# If workspace_id provided in query, check if it matches jwt or user is admin
if workspace_id:
if not jwt_params.ad and jwt_params.ap != workspace_id:
raise AuthenticationException("Unauthorized access to resource")
target_workspace_id = workspace_id
else:
# Use workspace_id from JWT
if not jwt_params.ap:
raise AuthenticationException("Workspace ID not found in query parameter or JWT")
target_workspace_id = jwt_params.ap
return await crud.get_workspace(db, workspace_id=target_workspace_id)
@router.post(
"/list",
response_model=Page[schemas.Workspace],
dependencies=[Depends(require_auth(admin=True))],
)
async def get_all_workspaces(
options: Optional[schemas.WorkspaceGet] = Body(
None, description="Filtering and pagination options for the workspaces list"
),
reverse: Optional[bool] = Query(
False, description="Whether to reverse the order of results"
),
db=db,
):
"""Get all Workspaces"""
filter_param = None
if options and hasattr(options, "filter"):
filter_param = options.filter
if filter_param == {}:
filter_param = None
return await paginate(
db,
await crud.get_all_workspaces(
reverse=reverse,
filter=filter_param,
),
)
@router.get(
"/name/{name}",
response_model=schemas.Workspace,
dependencies=[Depends(require_auth(admin=True))],
)
async def get_workspace_by_name(
name: str = Path(..., description="Name of the workspace to retrieve"),
db=db,
):
"""Get a Workspace by Name"""
# ResourceNotFoundException will be caught by global handler if workspace not found
workspace = await crud.get_workspace_by_name(db, name=name)
return workspace
@router.post(
"", response_model=schemas.Workspace, dependencies=[Depends(require_auth(admin=True))]
)
async def create_workspace(
workspace: schemas.WorkspaceCreate = Body(..., description="Workspace creation parameters"),
db=db,
):
"""Create a new Workspace"""
honcho_workspace = await crud.create_workspace(db, workspace=workspace)
return honcho_workspace
@router.get(
"/get_or_create/{name}",
response_model=schemas.Workspace,
dependencies=[Depends(require_auth(admin=True))],
)
async def get_or_create_workspace(
name: str = Path(..., description="Name of the workspace to get or create"),
db=db,
):
"""Get or Create a Workspace"""
try:
workspace = await crud.get_workspace_by_name(db=db, name=name)
return workspace
except ResourceNotFoundException:
# Workspace doesn't exist, create it
workspace = await create_workspace(db=db, workspace=schemas.WorkspaceCreate(name=name))
return workspace
@router.put(
"/{workspace_id}",
response_model=schemas.Workspace,
dependencies=[Depends(require_auth(app_id="workspace_id"))],
)
async def update_workspace(
workspace_id: str = Path(..., description="ID of the workspace to update"),
workspace: schemas.WorkspaceUpdate = Body(..., description="Updated workspace parameters"),
db=db,
):
"""Update a Workspace"""
# ResourceNotFoundException will be caught by global handler if workspace not found
honcho_workspace = await crud.update_workspace(db, workspace_id=workspace_id, workspace=workspace)
return honcho_workspace