FastAPI for ML Microservices with Pydantic v2

Why FastAPI for ML Microservices?

In 2026, deploying ML models in production requires three things: speed (latency < 200ms), rigorous validation (no garbage input), and resilience (fallback when the model is down). FastAPI + Pydantic v2 + Python 3.12 answers these requirements natively.

Unlike Flask (legacy, slow, no type hints), FastAPI offers:

• Native async/await: handle hundreds of concurrent requests on a single CPU core.
• Built-in Pydantic v2: schema-first validation, automatic serialization, precise error messages.
• Auto-generated OpenAPI docs: Swagger UI for free, no manual docstrings.
• Type hints: code intelligence in IDEs, catch bugs before runtime.
• Dependency injection: inject models, cache, logger, settings declaratively.

In Federico Calò's portfolio, the FastAPI ML microservice on VPS Hetzner serves 40+ RAG endpoints with average latency 85ms (p99 < 350ms) with zero downtime, handling 2M requests/month.

Pydantic v2: Models and Validators

Defining Request/Response Models

In Pydantic v2, a model is a class that inherits from BaseModel. The magic: type hints → automatic validation.

from pydantic import BaseModel, Field, field_validator
from typing import Optional
from enum import Enum

class EmbeddingRequest(BaseModel):
    """Embedding request: text → vector."""
    text: str = Field(..., min_length=1, max_length=2000, description="Text to embed")
    model: str = Field(default="sentence-transformers/all-MiniLM-L6-v2")
    normalize: bool = Field(default=True, description="Normalize vector?")

    @field_validator('text')
    @classmethod
    def text_not_only_spaces(cls, v):
        if not v.strip():
            raise ValueError("Text cannot be only spaces")
        return v.strip()

class EmbeddingResponse(BaseModel):
    """Embedding response."""
    embedding: list[float] = Field(..., description="Embedding vector (dim 384)")
    model: str
    dims: int
    computation_time_ms: float = Field(..., ge=0, description="Processing time")
  

Notes:

• Field(...) = required, Field(default=...) = optional.
• field_validator = custom validator executed after parsing.
• ge=0 = greater or equal (schema validation).

SettingsModel for Configuration

Pydantic v2 introduces BaseSettings to load environment variables with type hints and validation:

from pydantic_settings import BaseSettings
from pydantic import Field

class MLServiceSettings(BaseSettings):
    """ML microservice configuration."""
    ollama_url: str = Field(default="http://localhost:11434", description="Ollama local URL")
    ollama_model: str = Field(default="llama3.1:8b")
    embedding_model: str = Field(default="sentence-transformers/all-MiniLM-L6-v2")
    max_context_length: int = Field(default=2048, ge=256)
    health_check_interval: float = Field(default=30.0, gt=0)
    timeout_seconds: float = Field(default=60.0, gt=0)

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = MLServiceSettings()
  

Loads from .env or system variables. If OLLAMA_URL is invalid (e.g., non-URL), Pydantic raises ValidationError at startup — fail-fast.

Async/Await for Non-Blocking I/O

FastAPI is built on async. Every endpoint must be async def to leverage I/O multiplexing via Uvicorn (ASGI server).

import httpx
import asyncio

class OllamaClient:
    """Async client to on-prem Ollama."""

    def __init__(self, base_url: str, timeout: float = 60.0):
        self.base_url = base_url.rstrip('/')
        self.timeout = timeout
        self._client: Optional[httpx.AsyncClient] = None

    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=self.timeout)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._client:
            await self._client.aclose()

    async def generate(self, prompt: str, model: str = "llama3.1:8b") -> str:
        """Generate text via Ollama (streaming)."""
        if not self._client:
            raise RuntimeError("Client not initialized")

        response = await self._client.post(
            f"{self.base_url}/api/generate",
            json={"model": model, "prompt": prompt, "stream": False}
        )
        response.raise_for_status()
        return response.json()["response"]
  

With async with OllamaClient(...) as client: text = await client.generate(...), the Uvicorn thread doesn't block while waiting for Ollama — it can serve other requests in parallel.

Lifespan: Model Loading at Boot

The most critical section. With lifespan, you load the model once at startup (not per request). FastAPI v0.93+ uses @asynccontextmanager:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import asyncio

# Global state (singleton)
class ModelCache:
    embedding_model = None
    ollama_client = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # STARTUP: load models
    print("[STARTUP] Loading models...")

    try:
        # Load embedding model in-memory
        from sentence_transformers import SentenceTransformer
        ModelCache.embedding_model = SentenceTransformer(
            "sentence-transformers/all-MiniLM-L6-v2",
            device="cpu"  # or "cuda" if GPU available
        )
        print("[STARTUP] Embedding model loaded")

        # Create Ollama client
        ModelCache.ollama_client = OllamaClient(
            base_url=settings.ollama_url,
            timeout=settings.timeout_seconds
        )
        await ModelCache.ollama_client.__aenter__()
        print("[STARTUP] Ollama client connected")

    except Exception as e:
        print(f"[ERROR] Startup failed: {e}")
        raise

    yield  # App runs here

    # SHUTDOWN: cleanup
    print("[SHUTDOWN] Closing resources...")
    if ModelCache.ollama_client:
        await ModelCache.ollama_client.__aexit__(None, None, None)
    print("[SHUTDOWN] Cleanup complete")

app = FastAPI(
    title="ML Microservice",
    version="1.0.0",
    lifespan=lifespan
)
  

The lifespan ensures:

• Models loaded ONCE at boot (not per request).
• If loading fails, the app doesn't start (fail-fast).
• Orderly shutdown: close connections, free memory.

Dependency Injection

FastAPI's Depends() injects dependencies into endpoints declaratively:

from fastapi import FastAPI, Depends

def get_embedding_model():
    """Dependency: return loaded embedding model."""
    if ModelCache.embedding_model is None:
        raise RuntimeError("Embedding model not loaded")
    return ModelCache.embedding_model

def get_ollama_client():
    """Dependency: return Ollama client."""
    if ModelCache.ollama_client is None:
        raise RuntimeError("Ollama client unavailable")
    return ModelCache.ollama_client

@app.post("/embed", response_model=EmbeddingResponse)
async def embed(
    request: EmbeddingRequest,
    model = Depends(get_embedding_model),
    ollama_client = Depends(get_ollama_client)
):
    """Embedding endpoint with injected dependencies."""
    import time
    start = time.perf_counter()

    # Use model and ollama_client
    embedding = model.encode(request.text).tolist()

    elapsed_ms = (time.perf_counter() - start) * 1000

    return EmbeddingResponse(
        embedding=embedding,
        model=request.model,
        dims=len(embedding),
        computation_time_ms=elapsed_ms
    )
  

Benefits: testability (mock dependencies), reusability (same dependency in multiple endpoints), and error propagation (if model is None, FastAPI responds 500 with clear message).

Automatic Request/Response Validation

Pydantic v2 validates automatically:

# Valid request
POST /embed
{"text": "Hello world", "normalize": true}
→ 200 OK

# Invalid request (text too long)
POST /embed
{"text": "x" * 2001}
→ 422 Unprocessable Entity
{
  "detail": [
    {
      "type": "string_too_long",
      "loc": ["body", "text"],
      "msg": "String should have at most 2000 characters",
      "input": "xxxx..."
    }
  ]
}
  

No manual try/catch for input validation — Pydantic does it automatically.

Structured Error Handling

Define custom exceptions and a global exception handler:

from fastapi import HTTPException, status

class OllamaUnavailableError(Exception):
    """Ollama unavailable."""
    pass

class ModelInferenceError(Exception):
    """Error during inference."""
    pass

@app.exception_handler(OllamaUnavailableError)
async def handle_ollama_unavailable(request, exc):
    return JSONResponse(
        status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
        content={
            "error": "ollama_unavailable",
            "message": "Ollama AI provider temporarily unavailable",
            "retry_after_seconds": 30
        }
    )

@app.exception_handler(ModelInferenceError)
async def handle_inference_error(request, exc):
    return JSONResponse(
        status_code=status.HTTP_400_BAD_REQUEST,
        content={
            "error": "inference_failed",
            "message": str(exc),
            "request_id": request.headers.get("X-Request-ID", "unknown")
        }
    )
  

Health Check and Graceful Degradation

A health check endpoint that monitors the status of dependent services:

from enum import Enum
from datetime import datetime

class HealthStatus(str, Enum):
    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

class HealthCheck(BaseModel):
    status: HealthStatus
    timestamp: str
    services: dict[str, dict]
    version: str = "1.0.0"

@app.get("/health", response_model=HealthCheck)
async def health_check():
    """Health check with status of all dependent services."""
    import time

    services = {}
    overall_status = HealthStatus.HEALTHY

    # Check embedding model
    if ModelCache.embedding_model is None:
        services["embedding_model"] = {"status": "down"}
        overall_status = HealthStatus.UNHEALTHY
    else:
        services["embedding_model"] = {"status": "up"}

    # Check Ollama
    if ModelCache.ollama_client is None:
        services["ollama"] = {"status": "down", "message": "Client not initialized"}
        overall_status = HealthStatus.UNHEALTHY
    else:
        try:
            start = time.perf_counter()
            response = await ModelCache.ollama_client._client.get(
                f"{ModelCache.ollama_client.base_url}/api/tags",
                timeout=5.0
            )
            elapsed_ms = (time.perf_counter() - start) * 1000

            if response.status_code == 200:
                services["ollama"] = {
                    "status": "up",
                    "latency_ms": round(elapsed_ms, 2)
                }
            else:
                services["ollama"] = {
                    "status": "degraded",
                    "http_code": response.status_code
                }
                overall_status = HealthStatus.DEGRADED

        except Exception as e:
            services["ollama"] = {"status": "down", "error": str(e)}
            overall_status = HealthStatus.UNHEALTHY

    return HealthCheck(
        status=overall_status,
        timestamp=datetime.utcnow().isoformat(),
        services=services
    )
  

Graceful Degradation: Deterministic Fallback

If Ollama doesn't respond, the service returns a credible result (not crash):

class ChatRequest(BaseModel):
    query: str = Field(..., max_length=1000)
    system_prompt: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    model: str
    ollama_available: bool
    latency_ms: float

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """Chat endpoint with graceful degradation."""
    import time
    start = time.perf_counter()

    ollama_available = False
    response_text = ""

    try:
        # Try Ollama
        async with OllamaClient(settings.ollama_url, timeout=10.0) as client:
            response_text = await client.generate(
                prompt=request.query,
                model=settings.ollama_model
            )
            ollama_available = True

    except asyncio.TimeoutError:
        # Ollama timeout: fallback
        response_text = f"[FALLBACK] Ollama did not respond in timeout. Query: {request.query[:100]}..."
        ollama_available = False

    except Exception as e:
        # Any other error
        response_text = f"[FALLBACK] Service error: {type(e).__name__}"
        ollama_available = False

    elapsed_ms = (time.perf_counter() - start) * 1000

    return ChatResponse(
        response=response_text,
        model=settings.ollama_model,
        ollama_available=ollama_available,
        latency_ms=round(elapsed_ms, 2)
    )
  

Complete RAG Example: Embedding + Retrieval + Generation

A realistic RAG microservice combining everything:

import numpy as np
from typing import List

class DocumentStore:
    """In-memory document store with embeddings."""

    def __init__(self):
        self.docs: List[dict] = []
        self.embeddings: List[np.ndarray] = []

    def add_document(self, doc_id: str, text: str, embedding: np.ndarray):
        """Add document with embedding."""
        self.docs.append({"id": doc_id, "text": text})
        self.embeddings.append(embedding)

    async def search(self, query_embedding: np.ndarray, top_k: int = 3) -> List[dict]:
        """Retrieval: return top-k most similar documents."""
        if not self.embeddings:
            return []

        scores = [
            np.dot(query_embedding, emb) / (np.linalg.norm(query_embedding) * np.linalg.norm(emb) + 1e-8)
            for emb in self.embeddings
        ]

        top_indices = np.argsort(scores)[-top_k:][::-1]
        return [
            {"doc": self.docs[i], "score": float(scores[i])}
            for i in top_indices
            if scores[i] > 0.3  # Only relevant documents
        ]

# Global instance
document_store = DocumentStore()

class RAGRequest(BaseModel):
    query: str = Field(..., max_length=500)
    top_k: int = Field(default=3, ge=1, le=10)
    context_window: int = Field(default=2000, ge=256, le=4096)

class RAGResponse(BaseModel):
    query: str
    retrieved_docs: List[dict]
    generated_response: str
    model: str
    retrieval_latency_ms: float
    generation_latency_ms: float

@app.post("/rag", response_model=RAGResponse)
async def rag_search_and_generate(
    request: RAGRequest,
    embedding_model = Depends(get_embedding_model),
    ollama_client = Depends(get_ollama_client)
):
    """RAG: retrieval + generation."""
    import time

    # 1. Embed query
    retrieval_start = time.perf_counter()
    query_embedding = embedding_model.encode(request.query)

    # 2. Retrieve related documents
    retrieved = await document_store.search(query_embedding, top_k=request.top_k)
    retrieval_ms = (time.perf_counter() - retrieval_start) * 1000

    # 3. Build context
    context = "\n".join([
        f"[Doc {i}] {doc['doc']['text'][:200]}"
        for i, doc in enumerate(retrieved)
    ])

    # 4. Generate response via Ollama
    generation_start = time.perf_counter()
    rag_prompt = f"""
Context:
{context}

Question: {request.query}

Answer:"""

    try:
        generated = await ollama_client.generate(prompt=rag_prompt)
    except Exception as e:
        generated = f"[Generation error]: {str(e)}"

    generation_ms = (time.perf_counter() - generation_start) * 1000

    return RAGResponse(
        query=request.query,
        retrieved_docs=retrieved,
        generated_response=generated,
        model=settings.ollama_model,
        retrieval_latency_ms=round(retrieval_ms, 2),
        generation_latency_ms=round(generation_ms, 2)
    )
  

Async Testing with pytest-asyncio

import pytest
from fastapi.testclient import TestClient

@pytest.mark.asyncio
async def test_health_check():
    """Test health check when models are loaded."""
    async with client as c:
        response = await c.get("/health")
        assert response.status_code == 200
        data = response.json()
        assert data["status"] in ["healthy", "degraded"]
        assert "services" in data

@pytest.mark.asyncio
async def test_embed_validation():
    """Test Pydantic validation on invalid request."""
    async with client as c:
        response = await c.post("/embed", json={"text": ""})
        assert response.status_code == 422  # Unprocessable Entity
        errors = response.json()["detail"]
        assert any("String should have at most" in e.get("msg", "") for e in errors)
  

Performance Optimization

In Federico Calò's production environment (VPS Hetzner CCX13, 4 CPU cores):

• Embedding latency: 5-8ms (for text < 100 words).
• RAG retrieval latency: 2-3ms (cosine similarity on 10k docs).
• Ollama generation latency: 150-300ms (model llama3.1:8b, p99).
• Throughput: 40+ concurrent requests without stress.

Bottleneck: the LLM model (Ollama) — 85% of total latency. Solution: cache common query generations (Redis or in-memory).

FAQ

Q: Can I use FastAPI with a PostgreSQL database?

A: Yes, via SQLAlchemy async + Alembic. In the portfolio, all tool generators use SQLAlchemy with async connection pooling for fast queries (< 50ms).

Q: How do I handle rate limiting?

A: Use slowapi (FastAPI library) or implement custom middleware with Redis. In the ML microservice: 120 req/min per client IP (via X-Forwarded-For header).

Q: Is Pydantic v2 backward-compatible with v1?

A: No, significant breaking changes. Migration path: Config becomes model_config, parse_obj becomes model_validate. See official docs for complete list.

Q: How do I monitor the microservice in production?

A: Expose Prometheus metrics via prometheus-client: request latency, errors by type, memory usage. In the portfolio: Prometheus scrapes every 30s + Grafana dashboard for visualization.

Q: Do I need to use Docker?

A: Recommended but not required. In the portfolio: Docker Compose for dev, systemd unit for VPS production. Standard Dockerfile: Python 3.12 slim + pip install + CMD uvicorn.