Articles / MLOps
Serve a scikit-learn Model with FastAPI and Docker
Train an Iris classifier and serve it with FastAPI. Protect both prediction routes, validate finite inputs, bound batches and test a Docker container.
A prediction API needs a clear request contract, a trusted model artifact and a way to reject unauthorized calls. This walkthrough trains an Iris classifier, serves single predictions and batches, and packages the same example into Docker. It is a local teaching example, with no measured throughput or production-readiness claim.
Files and tested versions
Use Python 3.13 and the following direct dependencies in requirements.txt. The commands below use PowerShell. Keep training and serving dependencies aligned; lock transitive dependencies and the base-image digest for a release.
iris-api/
requirements.txt
train_model.py
test_api.py
app/
main.py
Dockerfile
fastapi==0.138.0
pydantic==2.13.4
uvicorn==0.49.0
scikit-learn==1.9.0
numpy==2.4.6
joblib==1.5.3
httpx==0.28.1
Train and save a trusted artifact
The scikit-learn introduction explains model evaluation. Here, a stratified holdout keeps each Iris class represented. The printed accuracy describes this small split, not deployment quality or future data.
# train_model.py
from pathlib import Path
import joblib
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
iris = load_iris()
X_train, X_test, y_train, y_test = train_test_split(
iris.data, iris.target, test_size=0.2, random_state=42, stratify=iris.target
)
model = RandomForestClassifier(n_estimators=100, random_state=42, n_jobs=1)
model.fit(X_train, y_train)
print(f"Holdout accuracy: {accuracy_score(y_test, model.predict(X_test)):.4f}")
model_dir = Path(__file__).resolve().parent / "model"
model_dir.mkdir(exist_ok=True)
joblib.dump(model, model_dir / "classifier.joblib")
joblib.dump(iris.target_names.tolist(), model_dir / "class_names.joblib")
Load only artifacts produced by your trusted training pipeline: joblib loading can execute code. Matching scikit-learn and dependency versions matters; copying an old artifact into a newly upgraded serving environment is unsupported. The Docker example below trains inside its own environment.
Protect both prediction routes
Create app/main.py. Inputs are positive, finite JSON numbers; strings, booleans and extra fields are rejected. A batch contains 1–128 records. Both endpoints use the same API-key dependency and prediction helper, preserving input order.
RandomForest converts inputs to float32 internally. A finite JSON number such as 1e100 exceeds that format and can otherwise turn inference into a server error. The schema applies its numeric upper limit to every feature; a real service also needs domain-specific ranges, not just numerical validity.
Configure API_KEY before importing the app. Missing or blank configuration prevents startup; missing and incorrect request keys return 401. Do not add a second handler for the same route to introduce authentication: attach the dependency to the original single and batch routes.
# app/main.py
from contextlib import asynccontextmanager
import os
from pathlib import Path
import secrets
from typing import Annotated
import joblib
import numpy as np
from fastapi import Depends, FastAPI, HTTPException, Security
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from fastapi.security import APIKeyHeader
from pydantic import BaseModel, ConfigDict, Field
EXPECTED_API_KEY = os.environ.get("API_KEY")
if not EXPECTED_API_KEY or not EXPECTED_API_KEY.strip():
raise RuntimeError("API_KEY must be configured before starting the API")
MODEL_DIR = Path(__file__).resolve().parents[1] / "model"
ml_models = {}
api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False)
def require_api_key(key: str | None = Security(api_key_header)):
if key is None or not secrets.compare_digest(
key.encode("utf-8"), EXPECTED_API_KEY.encode("utf-8")
):
raise HTTPException(
status_code=401, detail="Invalid API key",
headers={"WWW-Authenticate": "APIKey"},
)
MAX_FLOAT32 = float(np.finfo(np.float32).max)
class IrisFeatures(BaseModel):
model_config = ConfigDict(strict=True, extra="forbid", allow_inf_nan=False)
sepal_length: float = Field(gt=0, le=MAX_FLOAT32, description="Sepal length in cm")
sepal_width: float = Field(gt=0, le=MAX_FLOAT32, description="Sepal width in cm")
petal_length: float = Field(gt=0, le=MAX_FLOAT32, description="Petal length in cm")
petal_width: float = Field(gt=0, le=MAX_FLOAT32, description="Petal width in cm")
class PredictionResponse(BaseModel):
prediction: int
class_name: str
probabilities: list[float]
@asynccontextmanager
async def lifespan(app: FastAPI):
ml_models["classifier"] = joblib.load(MODEL_DIR / "classifier.joblib")
ml_models["class_names"] = joblib.load(MODEL_DIR / "class_names.joblib")
try:
yield
finally:
ml_models.clear()
app = FastAPI(title="Iris Classifier API", version="1.0.0", lifespan=lifespan)
@app.exception_handler(RequestValidationError)
async def invalid_request(request, error: RequestValidationError):
# Raw invalid values can contain NaN/infinity, which JSON cannot serialize.
details = [{"loc": item["loc"], "msg": item["msg"], "type": item["type"]}
for item in error.errors()]
return JSONResponse(status_code=422, content={"detail": details})
@app.get("/health")
def health_check():
if "classifier" not in ml_models or "class_names" not in ml_models:
raise HTTPException(status_code=503, detail="Model unavailable")
return {"status": "healthy", "model_loaded": True}
def predict_many(features_list: list[IrisFeatures]):
if "classifier" not in ml_models or "class_names" not in ml_models:
raise HTTPException(status_code=503, detail="Model unavailable")
matrix = np.array([[f.sepal_length, f.sepal_width, f.petal_length, f.petal_width]
for f in features_list])
model = ml_models["classifier"]
predictions = model.predict(matrix).tolist()
probabilities = model.predict_proba(matrix).tolist()
return [
PredictionResponse(
prediction=label, class_name=ml_models["class_names"][label],
probabilities=scores,
)
for label, scores in zip(predictions, probabilities)
]
@app.post("/predict", response_model=PredictionResponse,
dependencies=[Depends(require_api_key)])
def predict(features: IrisFeatures):
return predict_many([features])[0]
@app.post("/predict/batch", response_model=list[PredictionResponse],
dependencies=[Depends(require_api_key)])
def predict_batch(features_list: Annotated[list[IrisFeatures], Field(min_length=1, max_length=128)]):
return predict_many(features_list)
/health is public and returns 503 if the model is unavailable. Missing artifacts fail startup. The validation handler omits raw invalid input so NaN or infinity cannot turn a validation response into a JSON serialization error. Probabilities follow model.classes_—[setosa, versicolor, virginica] for this model—and are not automatically calibrated confidence.
Exercise the API without a running server
Create test_api.py. TestClient enters the lifespan, loads the actual saved model and sends requests to the application in process. These checks cover usable outputs, authentication on both endpoints and input boundaries.
# test_api.py
import os
from fastapi.testclient import TestClient
from app.main import app
sample = {"sepal_length": 5.1, "sepal_width": 3.5, "petal_length": 1.4, "petal_width": 0.2}
headers = {"X-API-Key": os.environ["API_KEY"]}
with TestClient(app) as client:
assert client.get("/health").status_code == 200
single = client.post("/predict", json=sample, headers=headers)
single.raise_for_status()
assert single.json()["class_name"] == "setosa"
batch = client.post("/predict/batch", json=[sample, sample], headers=headers)
batch.raise_for_status()
assert batch.json() == [single.json(), single.json()]
for route, payload in [("/predict", sample), ("/predict/batch", [sample])]:
assert client.post(route, json=payload).status_code == 401
assert client.post(route, json=payload, headers={"X-API-Key": "wrong"}).status_code == 401
assert client.post("/predict/batch", json=[], headers=headers).status_code == 422
assert client.post("/predict/batch", json=[sample] * 129, headers=headers).status_code == 422
assert client.post("/predict", json={**sample, "sepal_length": 0}, headers=headers).status_code == 422
for route, payload in [("/predict", {**sample, "sepal_length": 1e100}),
("/predict/batch", [{**sample, "sepal_length": 1e100}])]:
assert client.post(route, json=payload, headers=headers).status_code == 422
print("API contract checks passed")
From the project root, install, generate a local demo key, train and run the checks. The generated key is assigned to the environment rather than written into source or printed by the shell.
python -m venv .venv
& .\.venv\Scripts\python.exe -m pip install -r requirements.txt
$env:API_KEY = & .\.venv\Scripts\python.exe -c "import secrets; print(secrets.token_urlsafe(32))"
& .\.venv\Scripts\python.exe train_model.py
& .\.venv\Scripts\python.exe test_api.py
To explore the API over HTTP, run the following command and open the local API docs. Use Swagger's Authorize control with the configured key for prediction calls. Stop the server before starting the container on the same port.
& .\.venv\Scripts\python.exe -m uvicorn app.main:app --host 127.0.0.1 --port 8000
Build and test the container
The image installs the same direct versions, trains the toy model in that environment and runs as an unprivileged user. The key is supplied at runtime. Save this as Dockerfile:
FROM python:3.13-slim
ENV PYTHONDONTWRITEBYTECODE=1 PYTHONUNBUFFERED=1
ENV OMP_NUM_THREADS=1 OPENBLAS_NUM_THREADS=1 MKL_NUM_THREADS=1
WORKDIR /app
COPY requirements.txt .
RUN python -m pip install --no-cache-dir --only-binary=:all: -r requirements.txt
COPY train_model.py test_api.py ./
COPY app/ ./app/
RUN python train_model.py && useradd --create-home appuser
USER appuser
EXPOSE 8000
CMD ["python", "-m", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
The first run executes the same contract checks inside Docker, without network access or a persistent server. The second serves HTTP on the host's loopback interface; it runs in the foreground and is removed when stopped.
docker build -t iris-classifier:demo .
docker run --rm --network none --env API_KEY iris-classifier:demo python test_api.py
docker run --rm --env API_KEY -p 127.0.0.1:8000:8000 iris-classifier:demo
Before exposing a real service
A shared API key does not provide user identity, roles, tenant isolation or automatic rotation. A real deployment needs HTTPS, managed secrets, request-size and rate limits, artifact provenance, monitoring and measured resource capacity. Each worker loads its own model; adding processes consumes memory. The synchronous handlers allow FastAPI to run inference off the event loop, but do not make CPU inference faster.
The historical fast-ml-deploy repository uses /v1/iris/predict and a different payload. Its README and this complete example describe separate interfaces. Continue with pipeline orchestration and labeled quality/data drift monitoring.
Sources
Are you building an AI or data platform?
If you need to adapt this approach to your system, tell me about your data and the technical decision you need to resolve.