Serving with FastAPI
FastAPI wraps ML models in typed HTTP endpoints with automatic OpenAPI docs. Load the model once at startup, validate inputs with Pydantic, and expose health checks for orchestration.
Recipe
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
app = FastAPI()
model = joblib.load("model.joblib")
class PredictRequest(BaseModel):
features: list[float]
@app.post("/predict")
def predict(req: PredictRequest) -> dict:
pred = model.predict([req.features])
return {"prediction": pred.tolist()}Working Example
"""serving_fastapi.py - production-ready model API."""
from __future__ import annotations
from contextlib import asynccontextmanager
from typing import Any
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
MODEL_PATH = "models/churn_pipeline.joblib"
MODEL_VERSION = "1.2.0"
model: Any = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global model
model = joblib.load(MODEL_PATH)
yield
model = None
app = FastAPI(title="Churn Predictor", version=MODEL_VERSION, lifespan=lifespan)
class PredictRequest(BaseModel):
tenure: float = Field(ge=0)
monthly_charges: float = Field(ge=0)
total_charges: float = Field(ge=0)
class PredictResponse(BaseModel):
churn_probability: float
churn_prediction: bool
model_version: str
@app.get("/health")
def health() -> dict:
return {"status": "ok", "model_loaded": model is not None}
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
if model is None:
raise HTTPException(503, "Model not loaded")
features = np.array([[req.tenure, req.monthly_charges, req.total_charges]])
proba = model.predict_proba(features)[0][1]
return PredictResponse(
churn_probability=round(float(proba), 4),
churn_prediction=proba >= 0.5,
model_version=MODEL_VERSION,
)Run: uvicorn serving_fastapi:app --host 0.0.0.0 --port 8000
Gotchas
- Loading model per request - slow and memory-heavy. Fix: load once in lifespan/startup.
- No input validation - bad features crash predict. Fix: Pydantic models with constraints.
- No health check - orchestrator cannot detect failures. Fix:
/healthendpoint. - Wrong feature order - silent wrong predictions. Fix: named fields, not raw lists.
Alternatives
| Alternative | Use When | Don't Use When |
|---|---|---|
| FastAPI | Python models, custom logic | Max throughput (use Triton) |
| BentoML | Packaged model serving | Simple single-endpoint API |
| Flask | Legacy apps | New projects (use FastAPI) |
| AWS Lambda + container | Low traffic, serverless | GPU inference |
FAQs
Async or sync endpoints?
Sync is fine for CPU inference; async for I/O-bound LLM calls.How do I serve PyTorch?
Load model in lifespan; predict in thread pool for CPU-bound work.Batch predictions?
Accept list of requests; predict in batch for throughput.Authentication?
API key middleware or OAuth2 via FastAPI dependencies.Rate limiting?
slowapi or reverse proxy rate limits.Containerize?
Dockerfile with model artifact + uvicorn CMD.Multiple models?
Separate endpoints or model name in request body.GPU serving?
Load model to CUDA in lifespan; single process per GPU.Logging predictions?
Middleware logs input hash, prediction, latency (not raw PII).OpenAPI docs?
Auto-generated at /docs - disable in production if needed.Load from MLflow?
`mlflow.pyfunc.load_model("models:/name/Production")` in lifespan.Testing?
TestClient from fastapi.testclient for integration tests.Related
Stack versions: This page was written for Python 3.14.0 (stable 3.14, maintenance 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+, and uv 0.6+.