diff --git a/src/main.py b/src/main.py index 5c254825..7ee96c55 100644 --- a/src/main.py +++ b/src/main.py @@ -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") diff --git a/src/routers/apps.py b/src/routers/apps.py deleted file mode 100644 index 9d0443e4..00000000 --- a/src/routers/apps.py +++ /dev/null @@ -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 diff --git a/src/routers/collections.py b/src/routers/collections.py deleted file mode 100644 index ca1adeb2..00000000 --- a/src/routers/collections.py +++ /dev/null @@ -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"} diff --git a/src/routers/documents.py b/src/routers/documents.py deleted file mode 100644 index 127b9bca..00000000 --- a/src/routers/documents.py +++ /dev/null @@ -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") diff --git a/src/routers/messages.py b/src/routers/messages.py index 5ad4d939..e1dedf1d 100644 --- a/src/routers/messages.py +++ b/src/routers/messages.py @@ -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") ) ], ) diff --git a/src/routers/metamessages.py b/src/routers/metamessages.py deleted file mode 100644 index dc882ddc..00000000 --- a/src/routers/metamessages.py +++ /dev/null @@ -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 diff --git a/src/routers/peers.py b/src/routers/peers.py new file mode 100644 index 00000000..a42ff560 --- /dev/null +++ b/src/routers/peers.py @@ -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, + ), + ) diff --git a/src/routers/sessions.py b/src/routers/sessions.py index b7108251..51298376 100644 --- a/src/routers/sessions.py +++ b/src/routers/sessions.py @@ -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 diff --git a/src/routers/users.py b/src/routers/users.py deleted file mode 100644 index eef601eb..00000000 --- a/src/routers/users.py +++ /dev/null @@ -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 diff --git a/src/routers/workspaces.py b/src/routers/workspaces.py new file mode 100644 index 00000000..f9b73b7b --- /dev/null +++ b/src/routers/workspaces.py @@ -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