Pydantic contracts for ingest and query

Your API is only as honest as its contracts. Without Pydantic models, the ingest endpoint will happily accept malformed payloads and fail deep inside the embedding loop. We pin the shape at the boundary so every downstream layer gets clean data.

models.py
python
from pydantic import BaseModel, Field
from typing import List


class Document(BaseModel):
    id: str
    text: str


class IngestRequest(BaseModel):
    docs: List[Document] = Field(..., min_length=1)


class IngestResponse(BaseModel):
    ingested: int
    embedder: str  # 'ray' or 'sequential'


class QueryRequest(BaseModel):
    question: str
    top_k: int = 4


class QueryResponse(BaseModel):
    answer: str
    sources: List[dict]


class StatsResponse(BaseModel):
    document_count: int
    collection: str
    embedder: str

min_length=1 on docs rejects empty batches at the boundary. IngestResponse exposes which embedder ran so clients can see whether parallel Ray was available or the service fell back to a single process.

router.py
python
@router.post('/ingest', response_model=IngestResponse)
async def ingest(req: IngestRequest):
    try:
        result = rag_service.ingest(req.docs)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f'Ingest failed: {exc}')
    return IngestResponse(**result)


@router.post('/query', response_model=QueryResponse)
async def query(req: QueryRequest):
    if not req.question.strip():
        raise HTTPException(status_code=400, detail='question is required')
    try:
        result = await rag_service.query(req.question, top_k=req.top_k)
    except Exception as exc:
        raise HTTPException(status_code=500, detail=f'Query failed: {exc}')
    return QueryResponse(**result)


@router.get('/stats', response_model=StatsResponse)
async def stats():
    return StatsResponse(**rag_service.stats())

response_model is not decoration. FastAPI uses it to serialize, validate, and generate the OpenAPI schema. If your service layer returns a shape the model rejects, you find out in dev, not in production.

Quiz: Quiz

Loading practice…