agent-platform/services/knowledge-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 ServiceHealth

from app.application.services import KnowledgeApplicationService
from app.bootstrap.settings import KnowledgeServiceSettings
from app.db.session import get_db
from app.domain.repositories import (
    KnowledgeBaseRepository,
    KnowledgeChunkRepository,
    KnowledgeDocumentRepository,
)
from app.schemas.knowledge import (
    KnowledgeBaseCreateRequest,
    KnowledgeBaseResponse,
    KnowledgeBaseStatusUpdateRequest,
    KnowledgeChunkResponse,
    KnowledgeDocumentCreateRequest,
    KnowledgeDocumentIngestResponse,
    KnowledgeDocumentResponse,
    KnowledgeSearchRequest,
    KnowledgeSearchResultResponse,
)

router = APIRouter()


def get_knowledge_settings() -> KnowledgeServiceSettings:
    return KnowledgeServiceSettings()


def get_knowledge_application_service(
    db: Session = Depends(get_db),
    settings: KnowledgeServiceSettings = Depends(get_knowledge_settings),
) -> KnowledgeApplicationService:
    return KnowledgeApplicationService(
        settings=settings,
        base_repository=KnowledgeBaseRepository(db),
        document_repository=KnowledgeDocumentRepository(db),
        chunk_repository=KnowledgeChunkRepository(db),
    )


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


@router.post("/bases", response_model=KnowledgeBaseResponse)
def create_base(
    payload: KnowledgeBaseCreateRequest,
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> KnowledgeBaseResponse:
    return KnowledgeBaseResponse.from_entity(service.create_base(payload))


@router.get("/bases", response_model=list[KnowledgeBaseResponse])
def list_bases(
    tenant_id: str = Query(...),
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> list[KnowledgeBaseResponse]:
    return [
        KnowledgeBaseResponse.from_entity(item)
        for item in service.list_bases(tenant_id=tenant_id)
    ]


@router.patch("/bases/{knowledge_base_id}/status", response_model=KnowledgeBaseResponse)
def update_base_status(
    knowledge_base_id: str,
    payload: KnowledgeBaseStatusUpdateRequest,
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> KnowledgeBaseResponse:
    entity = service.update_base_status(
        knowledge_base_id=knowledge_base_id,
        payload=payload,
    )
    if entity is None:
        raise HTTPException(
            status_code=404,
            detail=f"knowledge base not found: {knowledge_base_id}",
        )
    return KnowledgeBaseResponse.from_entity(entity)


@router.post("/documents", response_model=KnowledgeDocumentIngestResponse)
def create_document(
    payload: KnowledgeDocumentCreateRequest,
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> KnowledgeDocumentIngestResponse:
    try:
        document, chunks = service.create_document(payload)
    except ValueError as exc:
        raise HTTPException(status_code=422, detail=str(exc)) from exc
    return KnowledgeDocumentIngestResponse(
        document=KnowledgeDocumentResponse.from_entity(document),
        chunks=[KnowledgeChunkResponse.from_entity(item) for item in chunks],
    )


@router.get("/documents", response_model=list[KnowledgeDocumentResponse])
def list_documents(
    tenant_id: str = Query(...),
    knowledge_base_id: str = Query(...),
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> list[KnowledgeDocumentResponse]:
    return [
        KnowledgeDocumentResponse.from_entity(item)
        for item in service.list_documents(
            tenant_id=tenant_id,
            knowledge_base_id=knowledge_base_id,
        )
    ]


@router.post("/search", response_model=list[KnowledgeSearchResultResponse])
def search(
    payload: KnowledgeSearchRequest,
    service: KnowledgeApplicationService = Depends(get_knowledge_application_service),
) -> list[KnowledgeSearchResultResponse]:
    return [
        KnowledgeSearchResultResponse(
            chunk=KnowledgeChunkResponse.from_entity(chunk),
            document=KnowledgeDocumentResponse.from_entity(document),
            score=score,
            score_json=score_json,
        )
        for chunk, document, score, score_json in service.search(payload)
    ]