agent-platform/services/memory-service/app/api/routes.py

from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy import text
from sqlalchemy.orm import Session

from core_domain import MemoryScopeType, MemoryStatus, ServiceHealth

from app.application.services import MemoryApplicationService
from app.db.session import get_db
from app.domain.repositories import MemoryItemRepository
from app.schemas.memory import (
    MemoryCreateRequest,
    MemoryResponse,
    MemorySearchRequest,
    MemorySearchResultResponse,
    MemoryStatusUpdateRequest,
)

router = APIRouter()


def get_memory_application_service(db: Session = Depends(get_db)) -> MemoryApplicationService:
    return MemoryApplicationService(memory_repository=MemoryItemRepository(db))


@router.get("/health", response_model=ServiceHealth)
def health_check(db: Session = Depends(get_db)) -> ServiceHealth:
    db.execute(text("SELECT 1"))
    return ServiceHealth(service="memory-service", status="ok", database="ok")


@router.post("", response_model=MemoryResponse)
def create_memory(
    payload: MemoryCreateRequest,
    service: MemoryApplicationService = Depends(get_memory_application_service),
) -> MemoryResponse:
    entity = service.create_memory(payload)
    return MemoryResponse.from_entity(entity)


@router.get("", response_model=list[MemoryResponse])
def list_memories(
    tenant_id: str = Query(...),
    scope_type: MemoryScopeType | None = Query(default=None),
    scope_id: str | None = Query(default=None),
    status: MemoryStatus | None = Query(default="active"),
    limit: int = Query(default=100, ge=1, le=500),
    service: MemoryApplicationService = Depends(get_memory_application_service),
) -> list[MemoryResponse]:
    return [
        MemoryResponse.from_entity(item)
        for item in service.list_memories(
            tenant_id=tenant_id,
            scope_type=scope_type,
            scope_id=scope_id,
            status=status,
            limit=limit,
        )
    ]


@router.post("/search", response_model=list[MemorySearchResultResponse])
def search_memories(
    payload: MemorySearchRequest,
    service: MemoryApplicationService = Depends(get_memory_application_service),
) -> list[MemorySearchResultResponse]:
    return [
        MemorySearchResultResponse.from_entity(item, score=score)
        for item, score in service.search_memories(payload)
    ]


@router.patch("/{memory_id}/status", response_model=MemoryResponse)
def update_memory_status(
    memory_id: str,
    payload: MemoryStatusUpdateRequest,
    service: MemoryApplicationService = Depends(get_memory_application_service),
) -> MemoryResponse:
    entity = service.update_memory_status(memory_id=memory_id, payload=payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=f"memory not found: {memory_id}")
    return MemoryResponse.from_entity(entity)