Module 09 Intermediate 22 min read

Building REST APIs

Wrap your model in a production-ready FastAPI service with Pydantic validation, batch endpoints, health checks, and automated tests.

Updated 2025 · Edit on GitHub

Why Build an API?

A saved model file is just bytes on disk. To make predictions useful — for a web app, a mobile app, another service — you need to wrap the model in an API (Application Programming Interface) that accepts HTTP requests with input data and returns predictions as JSON.

FastAPI — Modern, Fast, Auto-Documented

FastAPI is the modern standard for ML APIs. It uses Python type hints for automatic request validation, generates interactive docs, and is built on ASGI for async performance.

Python
# pip install fastapi uvicorn pydantic joblib scikit-learn
# File: app.py

import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List
import uvicorn

# ── Load model once at startup (not on every request)
pipe = joblib.load("models/v1/pipeline.joblib")

# ── FastAPI app
app = FastAPI(
    title="ZeroToML Breast Cancer API",
    description="Predicts malignant/benign from 30 tumour features",
    version="1.0.0",
)

# ── Request schema (Pydantic validates types automatically)
class PredictRequest(BaseModel):
    features: List[float] = Field(
        ...,
        min_length=30, max_length=30,
        example=[17.99, 10.38, 122.8, 1001.0, 0.1184, 0.2776,
                 0.3001, 0.1471, 0.2419, 0.07871, 1.095, 0.9053,
                 8.589, 153.4, 0.006399, 0.04904, 0.05373, 0.01587,
                 0.03003, 0.006193, 25.38, 17.33, 184.6, 2019.0,
                 0.1622, 0.6656, 0.7119, 0.2654, 0.4601, 0.1189]
    )

# ── Response schema
class PredictResponse(BaseModel):
    prediction:      int
    label:           str
    probability:     float
    confidence:      str

# ── Health check endpoint
@app.get("/health")
async def health():
    return {{"status": "ok", "model_loaded": pipe is not None}}

# ── Prediction endpoint
@app.post("/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
    try:
        X = np.array(request.features).reshape(1, -1)
        pred      = int(pipe.predict(X)[0])
        prob      = float(pipe.predict_proba(X)[0][pred])
        label     = "Benign" if pred == 1 else "Malignant"
        confidence = "High" if prob > 0.85 else "Medium" if prob > 0.65 else "Low"
        return PredictResponse(
            prediction=pred, label=label,
            probability=round(prob, 4), confidence=confidence
        )
    except Exception as e:
        raise HTTPException(status_code=422, detail=str(e))

# ── Batch prediction endpoint
@app.post("/predict/batch")
async def predict_batch(requests: List[PredictRequest]):
    X = np.array([r.features for r in requests])
    preds = pipe.predict(X).tolist()
    probs = pipe.predict_proba(X).max(axis=1).tolist()
    return [{{"prediction": p, "probability": round(pr, 4)}}
            for p, pr in zip(preds, probs)]

if __name__ == "__main__":
    uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)
Bash
# Start the server
uvicorn app:app --reload --host 0.0.0.0 --port 8000

# API docs auto-generated at:
# http://localhost:8000/docs   (Swagger UI)
# http://localhost:8000/redoc  (ReDoc)

# Test with curl
curl -X POST http://localhost:8000/predict   -H "Content-Type: application/json"   -d '{{"features": [17.99,10.38,122.8,1001.0,0.1184,0.2776,0.3001,0.1471,0.2419,0.07871,1.095,0.9053,8.589,153.4,0.006399,0.04904,0.05373,0.01587,0.03003,0.006193,25.38,17.33,184.6,2019.0,0.1622,0.6656,0.7119,0.2654,0.4601,0.1189]}}'

# Test with Python
python -c "
import requests
r = requests.post('http://localhost:8000/predict',
    json={{'features': [17.99]*30}})
print(r.json())
"

Flask — Lightweight Alternative

Python
# pip install flask joblib
# File: flask_app.py

import joblib
import numpy as np
from flask import Flask, request, jsonify

app   = Flask(__name__)
pipe  = joblib.load("models/v1/pipeline.joblib")

@app.route("/health", methods=["GET"])
def health():
    return jsonify({{"status": "ok"}})

@app.route("/predict", methods=["POST"])
def predict():
    data = request.get_json(force=True)
    if "features" not in data:
        return jsonify({{"error": "Missing 'features' key"}}), 400
    X    = np.array(data["features"]).reshape(1, -1)
    pred = int(pipe.predict(X)[0])
    prob = float(pipe.predict_proba(X)[0][pred])
    return jsonify({{
        "prediction":  pred,
        "label":       "Benign" if pred == 1 else "Malignant",
        "probability": round(prob, 4),
    }})

if __name__ == "__main__":
    app.run(debug=True, host="0.0.0.0", port=5000)

API Testing

Python
# File: test_api.py
import pytest
from fastapi.testclient import TestClient
from app import app          # import FastAPI app

client = TestClient(app)

def test_health():
    r = client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"

def test_predict_valid():
    r = client.post("/predict", json={{"features": [17.99]*30}})
    assert r.status_code == 200
    data = r.json()
    assert "prediction"  in data
    assert data["prediction"] in [0, 1]
    assert 0 <= data["probability"] <= 1

def test_predict_wrong_length():
    r = client.post("/predict", json={{"features": [1.0]*10}})  # wrong length
    assert r.status_code == 422    # Pydantic validation error

# Run: pytest test_api.py -v

Summary

  • FastAPI: recommended for new projects. Automatic validation via Pydantic, auto-generated docs, async-native.
  • Flask: simpler, more flexible, vast ecosystem. Good for quick prototypes.
  • Load the model once at startup — not on every request.
  • Always validate inputs (feature count, types) before passing to the model.
  • Write tests with TestClient. Test happy paths and edge cases.