A tabular classifier with a FastAPI endpoint
Train a scikit-learn pipeline on the breast cancer dataset, save it with joblib, serve it from a FastAPI app with /predict and /health, and test the app in-process before you ever start a server.
| Runs on | Laptop CPU, any OS, under 5 minutes |
|---|---|
| Time | about 2 hours |
| Stack | sklearn, python |
| Requirements | fastapi, uvicorn, joblib |
What you will build
This project takes the shortest honest path from a trained model to a working HTTP service. You will fit a scikit-learn pipeline (a scaler followed by logistic regression) on the bundled breast cancer dataset, save the whole pipeline to one file with joblib, and wrap it in a FastAPI application with two routes. GET /health says whether the model is loaded. POST /predict takes 30 numbers and returns a class label with a probability. Then you will test the application in-process with FastAPI’s TestClient, which exercises the real routing, validation and serialisation code without opening a port, and only after that start uvicorn and call it with curl.
It is a small project by design. The model is deliberately boring so that the serving side gets all the attention. If you have finished the beginner lessons on evaluation and moving from notebook to service, this is the first thing to build with them. The pattern (pipeline file, metadata file, lifespan loader, typed request and response, in-process tests) carries over unchanged to gradient-boosted trees, PyTorch models and anything else that can be loaded once and called many times.
What you need
- Python 3.10 or later with
scikit-learn,numpyandjoblib. Any recent version works. We tested with scikit-learn 1.9.1. fastapi,uvicornandhttpx, installed withpip install fastapi uvicorn httpx. FastAPI bringspydanticandstarlettewith it.httpxis whatfastapi.testclientuses to make in-process requests, and the plainfastapipackage does not install it (fastapi[standard]does).curlor any HTTP client for the final check.- No GPU, no network after the installs, and no downloads:
load_breast_cancerships inside scikit-learn.
Every Python block below was run from an empty directory with the Field Guide’s pipeline environment on an Apple Silicon Mac mini. Timings are given with each block.
Design
The service has three parts and a clear boundary between them.
Training is a separate program from serving. train.py fits the pipeline and writes two files. model.joblib holds the scaler and the classifier together, so the server can never apply the wrong scaling. model_meta.json holds the feature names in order, the class names and the test accuracy, so the server can validate inputs and label outputs without importing the training data. The server never trains anything.
The model loads once, at startup. FastAPI’s lifespan hook runs before the first request and after the last one. Loading there, rather than inside the route, means the first request is as fast as the thousandth, and a missing model file fails the process at start instead of returning errors under load.
Pydantic models define the contract. A PredictRequest with a list[float] field means a string, a nested object or a missing key is rejected with HTTP 422 before your code runs. The feature count is a semantic check that pydantic cannot do, so the route does it and raises the same 422 with a plain message.
Tests run against the app object, not a server. TestClient drives the ASGI application directly. Used as a context manager it also runs the lifespan, so the test covers startup. This is what you run in continuous integration. The uvicorn step at the end is for seeing it live.
One trade-off to be aware of: joblib files are Python pickles. They are only safe to load from a source you trust, and they tie you to the scikit-learn version that wrote them. Both are fine for a model you trained yourself an hour ago. For long-lived services, pin the library version in your requirements and record it in the metadata file.
Code
Train and save the pipeline
Save this as train.py. The dataset has 569 rows of 30 measurements from cell images, labelled malignant or benign. A stratified 80/20 split keeps the class balance the same in both halves. The Pipeline matters: it means predict_proba on raw inputs standardises them with the training statistics first.
# train.py: fit a scikit-learn pipeline on the breast cancer data and save it.
import json
import joblib
import numpy as np
from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
SEED = 0
data = load_breast_cancer()
X, y = data.data, data.target
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=SEED, stratify=y
)
pipe = Pipeline([
("scale", StandardScaler()),
("clf", LogisticRegression(max_iter=1000, random_state=SEED)),
])
pipe.fit(X_train, y_train)
acc = pipe.score(X_test, y_test)
joblib.dump(pipe, "model.joblib")
# The API needs the feature names and the class labels, so save them too.
meta = {
"features": list(data.feature_names),
"classes": list(data.target_names),
"test_accuracy": round(float(acc), 4),
}
with open("model_meta.json", "w") as f:
json.dump(meta, f, indent=2)
print(f"train rows: {len(X_train)}, test rows: {len(X_test)}")
print(f"test accuracy: {acc:.4f}")
print(f"saved model.joblib and model_meta.json ({len(meta['features'])} features)")
Expected output, about 1 s of compute (the first run took 10 s on our Mac mini, nearly all of it importing scikit-learn from a network volume):
train rows: 455, test rows: 114
test accuracy: 0.9825
saved model.joblib and model_meta.json (30 features)
The model file is about 2 kB. Logistic regression stores one weight per feature plus a bias, and the scaler stores a mean and a standard deviation per feature.
The application
Save this as app.py in the same directory. It is shown as a snippet because it is a file to run with uvicorn, not a script that prints something.
# snippet
# app.py: serve the saved pipeline over HTTP. Run with: uvicorn app:app --port 8000
import json
from contextlib import asynccontextmanager
from pathlib import Path
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
MODEL_PATH = Path("model.joblib")
META_PATH = Path("model_meta.json")
state: dict = {"model": None, "meta": None}
@asynccontextmanager
async def lifespan(app: FastAPI):
# Runs once at startup, before the first request is served.
state["model"] = joblib.load(MODEL_PATH)
state["meta"] = json.loads(META_PATH.read_text())
yield
state["model"] = None
app = FastAPI(title="Tumour classifier", version="1.0", lifespan=lifespan)
class PredictRequest(BaseModel):
features: list[float] = Field(..., description="30 values in model_meta.json order")
class PredictResponse(BaseModel):
label: str
probability: float
model_version: str
@app.get("/health")
def health() -> dict:
ready = state["model"] is not None
return {"status": "ok" if ready else "loading", "model_loaded": ready}
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
meta = state["meta"]
expected = len(meta["features"])
if len(req.features) != expected:
raise HTTPException(
status_code=422,
detail=f"expected {expected} features, got {len(req.features)}",
)
x = np.asarray(req.features, dtype=float).reshape(1, -1)
proba = state["model"].predict_proba(x)[0]
idx = int(np.argmax(proba))
return PredictResponse(
label=meta["classes"][idx],
probability=round(float(proba[idx]), 4),
model_version=app.version,
)
A few choices worth explaining. The routes are plain def, not async def. scikit-learn inference is synchronous CPU work, and FastAPI runs synchronous routes in a thread pool, so a slow prediction does not stall the event loop. The state dictionary is the simplest way to hand objects from the lifespan to the routes. response_model makes FastAPI validate what you return and publishes the shape in the generated OpenAPI document at /docs. The response carries a model_version so a client can tell which model answered, which matters the first time you deploy a second one.
Test it in-process
This block is runnable on its own. It repeats the training step if the model files are missing, defines the same application as app.py, and then calls it through TestClient. In a real project the test file would from app import app instead of repeating the code. The repetition here is so the block runs from an empty directory.
# Self-contained test: trains the model if needed, defines the same app, calls it in-process.
import json
from contextlib import asynccontextmanager
from pathlib import Path
import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field
from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
MODEL_PATH = Path("model.joblib")
META_PATH = Path("model_meta.json")
# --- Step 1 again, so this block works in an empty directory ---
data = load_breast_cancer()
if not MODEL_PATH.exists():
X_train, X_test, y_train, y_test = train_test_split(
data.data, data.target, test_size=0.2, random_state=0, stratify=data.target
)
pipe = Pipeline([("scale", StandardScaler()),
("clf", LogisticRegression(max_iter=1000, random_state=0))])
pipe.fit(X_train, y_train)
joblib.dump(pipe, MODEL_PATH)
META_PATH.write_text(json.dumps({
"features": list(data.feature_names),
"classes": list(data.target_names),
"test_accuracy": round(float(pipe.score(X_test, y_test)), 4),
}))
# --- The app, identical to app.py ---
state: dict = {"model": None, "meta": None}
@asynccontextmanager
async def lifespan(app: FastAPI):
state["model"] = joblib.load(MODEL_PATH)
state["meta"] = json.loads(META_PATH.read_text())
yield
state["model"] = None
app = FastAPI(title="Tumour classifier", version="1.0", lifespan=lifespan)
class PredictRequest(BaseModel):
features: list[float] = Field(..., description="30 values in model_meta.json order")
class PredictResponse(BaseModel):
label: str
probability: float
model_version: str
@app.get("/health")
def health() -> dict:
ready = state["model"] is not None
return {"status": "ok" if ready else "loading", "model_loaded": ready}
@app.post("/predict", response_model=PredictResponse)
def predict(req: PredictRequest) -> PredictResponse:
meta = state["meta"]
expected = len(meta["features"])
if len(req.features) != expected:
raise HTTPException(status_code=422,
detail=f"expected {expected} features, got {len(req.features)}")
x = np.asarray(req.features, dtype=float).reshape(1, -1)
proba = state["model"].predict_proba(x)[0]
idx = int(np.argmax(proba))
return PredictResponse(label=meta["classes"][idx],
probability=round(float(proba[idx]), 4),
model_version=app.version)
# --- Tests. The `with` runs the lifespan startup, as uvicorn would. ---
with TestClient(app) as client:
r = client.get("/health")
assert r.status_code == 200 and r.json()["model_loaded"], r.text
print("GET /health ->", r.json())
# Row 0 of the dataset is a known malignant case, row 19 a known benign one.
for row in (0, 19):
payload = {"features": data.data[row].tolist()}
r = client.post("/predict", json=payload)
assert r.status_code == 200, r.text
truth = data.target_names[data.target[row]]
print(f"POST /predict row {row:2d} -> {r.json()} (truth: {truth})")
r = client.post("/predict", json={"features": [1.0, 2.0, 3.0]})
print("POST /predict with 3 features ->", r.status_code, r.json()["detail"])
r = client.post("/predict", json={"features": "not a list"})
print("POST /predict with bad JSON type ->", r.status_code)
print("all checks passed")
Expected output, about 1.4 s on the Mac mini:
GET /health -> {'status': 'ok', 'model_loaded': True}
POST /predict row 0 -> {'label': 'malignant', 'probability': 1.0, 'model_version': '1.0'} (truth: malignant)
POST /predict row 19 -> {'label': 'benign', 'probability': 0.9328, 'model_version': '1.0'} (truth: benign)
POST /predict with 3 features -> 422 expected 30 features, got 3
POST /predict with bad JSON type -> 422
all checks passed
Rows 0 and 19 are both in the training set here, so the confident answers prove plumbing, not generalisation. The 98 % test accuracy printed by train.py is the number that speaks to generalisation. The two 422 cases are the ones that matter most in practice: one is caught by your code, the other by pydantic before your code runs.
Run
From the directory that holds train.py and app.py:
pip install scikit-learn joblib fastapi uvicorn httpx
python train.py
uvicorn app:app --host 127.0.0.1 --port 8000
uvicorn prints Application startup complete once the lifespan has loaded the model. Leave it running and use a second terminal for the checks below. Add --reload while editing app.py so the server restarts on every save. Do not use --reload in production, and bind to 0.0.0.0 only when you mean to accept connections from other machines.
Test
Health first:
curl -s http://127.0.0.1:8000/health
{"status":"ok","model_loaded":true}
Now a prediction. The 30 numbers below are row 19 of the dataset rounded to four decimals, a benign case:
curl -s -X POST http://127.0.0.1:8000/predict \
-H 'Content-Type: application/json' \
-d '{"features": [13.54, 14.36, 87.46, 566.3, 0.0978, 0.0813, 0.0666, 0.0478, 0.1885, 0.0577, 0.2699, 0.7886, 2.058, 23.56, 0.0085, 0.0146, 0.0239, 0.0132, 0.0198, 0.0023, 15.11, 19.26, 99.7, 711.2, 0.144, 0.1773, 0.239, 0.1288, 0.2977, 0.0726]}'
{"label":"benign","probability":0.9326,"model_version":"1.0"}
The probability differs from the in-process test in the fourth decimal place because the inputs were rounded. A short request should fail cleanly:
curl -s -X POST http://127.0.0.1:8000/predict \
-H 'Content-Type: application/json' -d '{"features": [1, 2, 3]}'
{"detail":"expected 30 features, got 3"}
Open http://127.0.0.1:8000/docs in a browser for the interactive OpenAPI page that FastAPI generates from the pydantic models. You can send requests from there too.
Things to try
- Swap the classifier for
HistGradientBoostingClassifierfromsklearn.ensemble. Nothing inapp.pychanges, which is the point of saving the whole pipeline. Compare the test accuracy and the file size. - Return the full probability vector and the feature names from a new
GET /schemaroute, read frommodel_meta.json, so a client can build a request without reading your source. - Add a
POST /predict/batchroute that accepts a list of rows. Measure the throughput of one request with 1,000 rows against 1,000 single-row requests usingTestClientandtime.perf_counter. - Record
sklearn.__version__in the metadata at training time and refuse to start if the serving version differs. It is a one-line check that prevents a confusing class of errors. - Write the tests as proper
pytestfunctions thatfrom app import appand use a fixture for the client. Then run them in a GitHub Actions workflow on every push. - Put the app in a container with a
python:3.12-slimbase image and the two model files copied in. The notebook to service lesson covers the Dockerfile.
Summary
- Save the whole pipeline, not the bare classifier, so preprocessing can never drift from the model.
- Keep a small metadata file beside the model with feature names, class names and the score, and let the server validate against it.
- Load the model once in a
lifespanhook, not inside the route. - Typed pydantic models give you validation, documentation and clear 422 errors for free. Add the semantic checks they cannot do.
- Test with
TestClientin-process first. uvicorn and curl are for confirming what the tests already proved.