Y

MLOps · Intermediate

Serve a model with FastAPI, and test it

4 min read

Why it matters

MLOps appeared in 58% of AI/ML Engineer postings we read, and "deploy models to production" is in most of them. Interviewers ask how you would get a model from a notebook to a service that other teams call, and what you would check before trusting it. This lesson builds that service in about 80 lines.

The concept

A minimal, production-shaped model service has:

  1. A saved pipeline, preprocessing and model together, so serving applies exactly the training transformations.

  2. Metadata next to it: version, training date, feature list, evaluation score.

  3. Input validation at the edge, rejecting impossible values before they reach the model.

  4. Endpoints: /health for load balancers and /score for predictions, both reporting the model version.

  5. Tests that run in CI: health, a known input, and rejection of bad input.

Then package it in a container (see the Docker shelf) and deploy it behind a load balancer.

Worked example: the term deposit model

Train on the UCI Bank Marketing data (CC BY 4.0), using only information known before a call:

train.py
"""Train the subscription model and save it with its metadata."""
import json
from datetime import date

import joblib
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder

FEATURES = ["age", "job", "marital", "education", "default", "balance", "housing",
            "loan", "contact", "campaign", "pdays", "previous", "poutcome"]
CATEGORICAL = ["job", "marital", "education", "default", "housing", "loan", "contact", "poutcome"]

bank = pd.read_csv("bank-full.csv", sep=";")
X, y = bank[FEATURES], (bank["y"] == "yes").astype(int)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

model = Pipeline([
    ("prep", ColumnTransformer([
        ("cat", OneHotEncoder(handle_unknown="ignore", sparse_output=False), CATEGORICAL),
    ], remainder="passthrough")),
    ("gbm", HistGradientBoostingClassifier(random_state=42)),
]).fit(X_train, y_train)

auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
joblib.dump(model, "model.joblib")
json.dump({"version": "2026.09.1", "trained_on": date.today().isoformat(), "features": FEATURES,
           "test_roc_auc": round(auc, 3), "rows": len(X_train)}, open("model.json", "w"), indent=1)
print("saved model, test ROC AUC", round(auc, 3))

It prints a test ROC AUC of about 0.77. Now the service:

app.py
"""Serve subscription scores over HTTP."""
import json
from typing import Literal

import joblib
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel, Field

model = joblib.load("model.joblib")
meta = json.load(open("model.json"))
app = FastAPI(title="Term deposit propensity", version=meta["version"])


class Customer(BaseModel):
    age: int = Field(ge=18, le=100)
    job: str
    marital: Literal["married", "single", "divorced"]
    education: Literal["primary", "secondary", "tertiary", "unknown"]
    default: Literal["yes", "no"]
    balance: int
    housing: Literal["yes", "no"]
    loan: Literal["yes", "no"]
    contact: Literal["cellular", "telephone", "unknown"]
    campaign: int = Field(ge=1)
    pdays: int
    previous: int = Field(ge=0)
    poutcome: Literal["success", "failure", "other", "unknown"]


@app.get("/health")
def health() -> dict:
    return {"status": "ok", "model_version": meta["version"]}


@app.post("/score")
def score(customers: list[Customer]) -> dict:
    frame = pd.DataFrame([c.model_dump() for c in customers])[meta["features"]]
    probabilities = model.predict_proba(frame)[:, 1]
    return {"model_version": meta["version"], "scores": [round(float(p), 4) for p in probabilities]}

And the tests:

test_app.py
from fastapi.testclient import TestClient

from app import app

client = TestClient(app)
customer = {"age": 35, "job": "management", "marital": "married", "education": "tertiary",
            "default": "no", "balance": 1500, "housing": "no", "loan": "no", "contact": "cellular",
            "campaign": 1, "pdays": 90, "previous": 2, "poutcome": "success"}


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


def test_score_returns_probability():
    body = client.post("/score", json=[customer]).json()
    assert 0 <= body["scores"][0] <= 1


def test_rejects_invalid_input():
    assert client.post("/score", json=[{**customer, "age": 7}]).status_code == 422
run.sh
pip install fastapi uvicorn scikit-learn pandas joblib pytest httpx
python train.py && pytest -q                    # 3 passed
uvicorn app:app --port 8000                      # then open http://localhost:8000/docs

For the customer above the service returns a score of about 0.71: a previous campaign success is a strong signal. The same customer with no previous contact scores about 0.05.

Common mistakes

  • Saving only the model, not the pipeline. Serving then re-implements preprocessing, and the two drift apart.

  • No input validation. A missing field or an age of 7 produces a confident nonsense score instead of an error.

  • Unversioned models. Every response should say which model produced it, so you can trace complaints.

  • Loading the model per request. Load once at startup; per-request loading is slow and wasteful.

Interview angle

"How would you deploy this model?" Walk through: pipeline plus metadata artefact, a validated API, tests, a container, a health check, gradual rollout (shadow or canary), and monitoring. Mention the feature-availability rule: only features known at prediction time.

Practice task

  1. Add a /model endpoint that returns the metadata.

  2. Write a Dockerfile for the service and run it locally.

  3. Log each request's model version, latency and score (not personal data) as one JSON line.

Further reading