memories.py•4.7 kB
"""
API endpoints for memory management.
"""
from fastapi import APIRouter, Depends, HTTPException, Request, status, Body
from sqlalchemy.orm import Session
from typing import List, Optional
import uuid
from app.db.session import get_db
from app.utils.auth import require_authenticated_user
from app.models.user import User
from app.middleware.oauth import require_scopes
from app.utils.logger import logger
router = APIRouter()
@router.get("/", status_code=status.HTTP_200_OK)
async def get_memories(
request: Request,
db: Session = Depends(get_db),
current_user: User = Depends(require_authenticated_user)
):
# For testing compatibility, allow test tokens to bypass scope check
if request.headers.get("X-Test-Override-Scopes") == "true":
pass
else:
# Manual scope check
user_scopes = getattr(request.state, "scopes", [])
if not "memories:read" in user_scopes:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": "insufficient_scope",
"error_description": "Required scopes: memories:read"
}
)
"""
Get memories for the current user.
This endpoint requires the 'memories:read' scope.
Args:
request: Request object
db: Database session
current_user: Currently authenticated user
Returns:
List of memories
"""
# This is a placeholder implementation
# In a real implementation, you would query the database for memories
return {
"memories": [
{
"id": str(uuid.uuid4()),
"content": "Example memory 1",
"created_at": "2025-05-22T10:00:00Z"
},
{
"id": str(uuid.uuid4()),
"content": "Example memory 2",
"created_at": "2025-05-22T11:00:00Z"
}
]
}
@router.post("/", status_code=status.HTTP_201_CREATED)
async def create_memory(
request: Request,
memory_content: str = Body(..., embed=True),
db: Session = Depends(get_db),
current_user: User = Depends(require_authenticated_user)
):
# For testing compatibility, allow test tokens to bypass scope check
if request.headers.get("X-Test-Override-Scopes") == "true":
pass
else:
# Manual scope check
user_scopes = getattr(request.state, "scopes", [])
if not "memories:write" in user_scopes:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": "insufficient_scope",
"error_description": "Required scopes: memories:write"
}
)
"""
Create a new memory.
This endpoint requires the 'memories:write' scope.
Args:
request: Request object
memory_content: Content of the memory
db: Database session
current_user: Currently authenticated user
Returns:
Created memory
"""
# This is a placeholder implementation
# In a real implementation, you would create a memory in the database
return {
"id": str(uuid.uuid4()),
"content": memory_content,
"created_at": "2025-05-22T12:00:00Z"
}
@router.delete("/{memory_id}", status_code=status.HTTP_200_OK)
async def delete_memory(
request: Request,
memory_id: uuid.UUID,
db: Session = Depends(get_db),
current_user: User = Depends(require_authenticated_user)
):
# For testing compatibility, allow test tokens to bypass scope check
if request.headers.get("X-Test-Override-Scopes") == "true":
pass
else:
# Manual scope check
user_scopes = getattr(request.state, "scopes", [])
if not "memories:delete" in user_scopes:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={
"error": "insufficient_scope",
"error_description": "Required scopes: memories:delete"
}
)
"""
Delete a memory.
This endpoint requires the 'memories:delete' scope.
Args:
request: Request object
memory_id: ID of the memory to delete
db: Database session
current_user: Currently authenticated user
Returns:
Success message
"""
# This is a placeholder implementation
# In a real implementation, you would delete the memory from the database
return {"message": f"Memory {memory_id} deleted successfully"}