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

from datetime import datetime
from typing import TypeVar
from uuid import uuid4

from core_domain import ServiceHealth
from core_shared import error_detail
from fastapi import APIRouter, Depends, HTTPException, Request
from sqlalchemy import text
from sqlalchemy.orm import Session

from app.application.services import MemoryApplicationService, build_memory_application_service
from app.bootstrap.settings import MemoryServiceSettings
from app.db.session import get_db
from app.schemas.memory import (
    ApiResponse,
    DeleteData,
    MemoryCreateRequestDto,
    MemoryDeleteRequestDto,
    MemoryDetailRequestDto,
    MemoryItemDto,
    MemoryListRequestDto,
    MemorySearchRequestDto,
    MemorySearchResultDto,
    MemoryStatusRequestDto,
    MemoryUpdateRequestDto,
    PageResult,
)

router = APIRouter()
T = TypeVar("T")


def ok(data: T) -> ApiResponse[T]:
    return ApiResponse(
        success=True,
        data=data,
        error=None,
        requestId=str(uuid4()),
        serverTime=datetime.utcnow())


def get_memory_application_service(
    request: Request,
    db: Session = Depends(get_db)) -> MemoryApplicationService:
    settings: MemoryServiceSettings = request.app.state.settings
    return build_memory_application_service(db=db, settings=settings)


@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("/list", response_model=ApiResponse[PageResult[MemoryItemDto]])
def list_memories_contract(
    payload: MemoryListRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[PageResult[MemoryItemDto]]:
    items, total = service.list_memories_contract(payload)
    return ok(PageResult[MemoryItemDto].from_items(
        items=[MemoryItemDto.from_entity(item) for item in items],
        total=total,
        page=payload.page,
        page_size=payload.pageSize))


@router.post("/create", response_model=ApiResponse[MemoryItemDto])
def create_memory_contract(
    payload: MemoryCreateRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[MemoryItemDto]:
    return ok(MemoryItemDto.from_entity(service.create_memory_from_contract(payload)))


@router.post("/detail", response_model=ApiResponse[MemoryItemDto])
def detail_memory_contract(
    payload: MemoryDetailRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[MemoryItemDto]:
    entity = service.get_memory(memory_id=payload.memoryId)
    if entity is None:
        raise HTTPException(status_code=404, detail=error_detail("error.memory.not_found", id=payload.memoryId))
    return ok(MemoryItemDto.from_entity(entity))


@router.post("/update", response_model=ApiResponse[MemoryItemDto])
def update_memory_contract(
    payload: MemoryUpdateRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[MemoryItemDto]:
    entity = service.update_memory(payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=error_detail("error.memory.not_found", id=payload.memoryId))
    return ok(MemoryItemDto.from_entity(entity))


@router.post("/status", response_model=ApiResponse[MemoryItemDto])
def update_memory_status_contract(
    payload: MemoryStatusRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[MemoryItemDto]:
    entity = service.update_memory_status(
        memory_id=payload.memoryId,
        payload=payload)
    if entity is None:
        raise HTTPException(status_code=404, detail=error_detail("error.memory.not_found", id=payload.memoryId))
    return ok(MemoryItemDto.from_entity(entity))


@router.post("/delete", response_model=ApiResponse[DeleteData])
def delete_memory_contract(
    payload: MemoryDeleteRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[DeleteData]:
    entity = service.delete_memory(memory_id=payload.memoryId)
    if entity is None:
        raise HTTPException(status_code=404, detail=error_detail("error.memory.not_found", id=payload.memoryId))
    return ok(DeleteData(deleted=True, memoryId=payload.memoryId))


@router.post("/search/query", response_model=ApiResponse[list[MemorySearchResultDto]])
def search_memories_contract(
    payload: MemorySearchRequestDto,
    service: MemoryApplicationService = Depends(get_memory_application_service)) -> ApiResponse[list[MemorySearchResultDto]]:
    return ok([
        MemorySearchResultDto(
            item=MemoryItemDto.from_entity(item),
            score=score,
            scoreDetails=score_json)
        for item, score, score_json in service.search_memories_contract(payload)
    ])