Project: Inference and Stats API

Introduction

FastAPI excels at exposing Python logic over HTTP—ML model inference, data transforms, batch statistics. This lightweight project accepts validated JSON, runs computation in-process, and optionally logs jobs with BackgroundTasks. No GPU or real model required; you simulate inference to learn the pattern used in production ML services.

Prerequisites

Project Goals

POST /api/v1/predict — JSON features → prediction JSON
POST /api/v1/stats/mean — list of numbers → mean/std (demo)
Pydantic validates input dimensions and ranges
BackgroundTasks writes audit line per prediction
Optional GET /api/v1/health for load balancers

Step 1: Schemas

schemas/inference.py:

python

Step 2: Fake Model Function

app/services/model.py:

python

def predict(features: list[float]) -> tuple[str, float]:
    # Stand-in for sklearn/torch/onnx runtime
    score = sum(features) / len(features)
    label = "positive" if score > 0.5 else "negative"
    return label, round(score, 4)

Replace with real model.predict() in production—load weights in lifespan once.

Step 3: Predict Endpoint

app/api/v1/endpoints/inference.py:

python

Register under /api/v1 → full path /api/v1/inference/predict or mount at /api/v1/predict by adjusting prefix.

Simpler path:

python

router = APIRouter(tags=["inference"])
 
@router.post("/predict", response_model=PredictResponse)
def predict_endpoint(...):
    ...

Include with prefix="/api/v1".

Step 4: Stats Endpoint

python

import statistics
 
@router.post("/stats/mean", response_model=StatsResponse)
def compute_stats(body: StatsRequest):
    mean = statistics.mean(body.values)
    std = statistics.pstdev(body.values) if len(body.values) > 1 else 0.0
    return StatsResponse(count=len(body.values), mean=mean, std=std)

Step 5: Load Real Model in Lifespan (Pattern)

python

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.model = load_model_from_disk("models/demo.pkl")
    yield
    app.state.model = None

Route uses request.app.state.model—avoid reloading file per request.

Step 6: Test

bash

curl -X POST http://127.0.0.1:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [0.2, 0.8, 0.5]}'

Invalid length → 422:

bash

curl -X POST http://127.0.0.1:8000/api/v1/predict \
  -H "Content-Type: application/json" \
  -d '{"features": [1, 2]}'

tests/test_inference.py:

python

def test_predict(client):
    r = client.post("/api/v1/predict", json={"features": [1.0, 1.0, 1.0]})
    assert r.status_code == 200
    assert "label" in r.json()
 
def test_predict_validation(client):
    assert client.post("/api/v1/predict", json={"features": []}).status_code == 422

Step 7: Production Notes

Concern	Approach
Heavy CPU	Worker queue (Celery), not sync route
Large batch	Separate batch endpoint + async job ID
Model GPU	Dedicated inference pod, scale horizontally
Timeouts	Nginx `proxy_read_timeout`, Gunicorn `--timeout`

See Flask background tasks for queue concepts.

Tip

Keep Inference Stateless

Store no session in memory between requests unless using app.state model weights loaded once.

Verification Checklist

Valid predict returns label and score
Invalid input returns 422 with detail
Background task logs prediction (check console/log file)
/docs documents request schema
pytest passes

FAQ

Blocking CPU in async route?

Use def route for CPU-bound fake model, or run_in_threadpool.

Upload model file?

Use file uploads—better load from object storage at startup.

Batch predict array?

features: list[list[float]] in schema—validate max batch size.

Auth on predict?

Add Depends(get_current_user) if internal-only API.

Monitor latency?

Log X-Process-Time-Ms from middleware chapter.

Real ML stack?

Swap predict() for ONNX Runtime, TorchServe, or Triton—HTTP contract stays same.