Model Deployment with FastAPI and Docker

Q: How do I rate-limit API requests?

Use the slowapi library for rate limiting — limiter = Limiter(key_func=get_remote_address) , e.g. capping requests at 100 per minute. This prevents abuse and overload.

A trained model sitting in Jupyter is worthless — its value is only unlocked once it's deployed to production.

1. What You'll Learn

FastAPI model serving: Pydantic validation, async prediction endpoints, and auto-generated Swagger docs
Model serialization: saving sklearn models with joblib/pickle and PyTorch models with torch.save
Docker containerization: writing Dockerfiles, multi-stage builds, and image optimization
docker-compose orchestration: model service + Redis cache + Nginx load balancing
Bob's prediction API: designing the POST /predict endpoint with sub-50ms latency per prediction

2. A Real ML Engineer's Story

(1) The Pain: A Notebook Model Can't Serve the Business

Bob trained an XGBoost model with R²=0.89, but it only ran inside his Jupyter Notebook. When the product manager asked, "Can the frontend call this?" Bob had no idea how to turn the model into an API. The gap between training and deployment is the biggest "last mile" problem in ML projects.

(2) The FastAPI + Docker Solution

FastAPI wraps the model in a REST API, and Docker containerizes it — any service can then call predictions over HTTP.

PYTHON

from fastapi import FastAPI
import joblib

app = FastAPI()
model = joblib.load("model.pkl")

@app.post("/predict")
def predict(features: PredictionInput):
    result = model.predict([features.dict()])
    return {"prediction": float(result[0])}

(3) The Payoff: Millions of Requests a Day After Launch

Once Bob deployed the model with FastAPI + Docker, the API latency stayed under 50ms while handling 1 million+ requests per day, callable from the frontend, CRM, and recommendation systems alike.

3. Model Serialization

(1) Saving and Loading Models

▶ Example: sklearn Model Serialization

PYTHON

import joblib
import pickle
from sklearn.ensemble import RandomForestRegressor
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
import numpy as np

# Train and save model
rng = np.random.default_rng(42)
X = rng.uniform(0, 100, (1000, 5))
y = 50 + 0.8 * X[:, 0] + 1.2 * X[:, 1] + rng.normal(0, 5, 1000)

pipe = Pipeline([
    ("scaler", StandardScaler()),
    ("model", RandomForestRegressor(n_estimators=100, random_state=42)),
])
pipe.fit(X, y)

# Save with joblib (recommended for sklearn)
joblib.dump(pipe, "salespredict_model.joblib", compress=3)

# Save with pickle (alternative)
with open("salespredict_model.pkl", "wb") as f:
    pickle.dump(pipe, f)

# Load and predict
loaded_model = joblib.load("salespredict_model.joblib")
sample = np.array([[50, 30, 20, 10, 5]])
prediction = loaded_model.predict(sample)
print(f"Prediction: {prediction[0]:.2f} thousand USD")

Output:

TEXT

# Executed successfully

Method	Best For	Pros	Cons
joblib	sklearn/numpy	Efficient for large arrays	Python only
pickle	Any Python object	Universal	Security risks, version compatibility
torch.save	PyTorch	Flexible (can save state_dict)	PyTorch only
mlflow.sklearn	sklearn	Versioning + metadata	Requires MLflow
ONNX	Cross-framework	Cross-language/platform	Complex conversion

▶ Example: Saving a PyTorch Model

PYTHON

import torch
import torch.nn as nn

# Save model state_dict (recommended)
class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.net = nn.Sequential(nn.Linear(5, 32), nn.ReLU(), nn.Linear(32, 1))

    def forward(self, x):
        return self.net(x)

model = SimpleModel()
torch.save(model.state_dict(), "pytorch_model.pt")

# Load
loaded = SimpleModel()
loaded.load_state_dict(torch.load("pytorch_model.pt", weights_only=True))
loaded.eval()

Output:

TEXT

# Function defined successfully

4. FastAPI Model Serving

(1) FastAPI Basics

▶ Example: A Complete Prediction API

PYTHON

# File: app.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
import joblib
import numpy as np
import time

app = FastAPI(title="SalesPredict API", version="1.0.0")

# Load model at startup
model = joblib.load("salespredict_model.joblib")

class PredictionInput(BaseModel):
    ad_spend_k: float = Field(..., ge=0, description="Ad spend in thousand USD")
    traffic_k: float = Field(..., ge=0, description="Traffic in thousands")
    category_electronics: float = Field(0, ge=0, le=1)
    category_clothing: float = Field(0, ge=0, le=1)
    is_promotion: float = Field(0, ge=0, le=1)

    model_config = {"json_schema_extra": {
        "example": {"ad_spend_k": 50, "traffic_k": 300,
                     "category_electronics": 1, "category_clothing": 0, "is_promotion": 1}
    }}

class PredictionOutput(BaseModel):
    predicted_revenue_k: float
    latency_ms: float

@app.get("/health")
def health_check():
    return {"status": "healthy", "model_loaded": model is not None}

@app.post("/predict", response_model=PredictionOutput)
def predict(input_data: PredictionInput):
    start = time.time()
    try:
        features = np.array([[input_data.ad_spend_k, input_data.traffic_k,
                               input_data.category_electronics,
                               input_data.category_clothing, input_data.is_promotion]])
        prediction = model.predict(features)[0]
        latency = (time.time() - start) * 1000
        return PredictionOutput(predicted_revenue_k=round(float(prediction), 2),
                                 latency_ms=round(latency, 2))
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.post("/predict_batch")
def predict_batch(inputs: list[PredictionInput]):
    features = np.array([[d.ad_spend_k, d.traffic_k, d.category_electronics,
                           d.category_clothing, d.is_promotion] for d in inputs])
    predictions = model.predict(features)
    return {"predictions": [round(float(p), 2) for p in predictions]}

Output:

TEXT

# Function defined successfully

(2) Running the FastAPI Service

BASH

# Install dependencies
pip install fastapi uvicorn joblib scikit-learn

# Run the API server
uvicorn app:app --host 0.0.0.0 --port 8000 --reload

# Test with curl
curl -X POST http://localhost:8000/predict \
  -H "Content-Type: application/json" \
  -d '{"ad_spend_k": 50, "traffic_k": 300, "category_electronics": 1, "category_clothing": 0, "is_promotion": 1}'

# Access Swagger UI: http://localhost:8000/docs

FastAPI Feature	Description
Pydantic validation	Automatically validates input types and ranges
Swagger UI	Auto-generates interactive docs (/docs)
Type hints	Auto-generates response models
Async support	async/await for high concurrency
Exception handling	Standard error codes via HTTPException

5. Docker Containerization

(1) Writing a Dockerfile

▶ Example: The SalesPredict Docker Image

DOCKERFILE

# Stage 1: Build dependencies
FROM python:3.11-slim AS builder

WORKDIR /build
COPY requirements.txt .
RUN pip install --no-cache-dir --prefix=/install -r requirements.txt

# Stage 2: Runtime (smaller image)
FROM python:3.11-slim

WORKDIR /app

# Copy installed packages from builder
COPY --from=builder /install /usr/local

# Copy application code and model
COPY app.py .
COPY salespredict_model.joblib .

# Non-root user for security
RUN useradd -m appuser
USER appuser

EXPOSE 8000

HEALTHCHECK --interval=30s --timeout=5s \
  CMD curl -f http://localhost:8000/health || exit 1

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

TEXT

# requirements.txt
fastapi==0.109.0
uvicorn==0.27.0
joblib==1.3.2
scikit-learn==1.4.0
numpy==1.26.4
pydantic==2.5.0

(2) Docker Commands

BASH

# Build image
docker build -t salespredict-api:latest .

# Run container
docker run -d -p 8000:8000 --name salespredict salespredict-api:latest

# Test
curl http://localhost:8000/health

# View logs
docker logs salespredict

# Stop and remove
docker stop salespredict && docker rm salespredict

Dockerfile Optimization	Effect
Multi-stage build	Shrinks the image from 1.5GB to 200MB
slim base image	Removes unnecessary system packages
.dockerignore	Excludes large files like .git/data
Non-root user	Security hardening
HEALTHCHECK	Container health checks

6. docker-compose Orchestration

A complete deployment architecture wires all the components together — the API service, cache, load balancer, and monitoring form a full end-to-end pipeline:

graph TB
    CLIENT[Client / Frontend] --> NGINX[Nginx<br/>Rate Limit + LB]
    NGINX --> API1[FastAPI Worker 1]
    NGINX --> API2[FastAPI Worker 2]
    API1 --> REDIS[(Redis Cache<br/>LRU 256MB)]
    API2 --> REDIS
    API1 --> MODEL[Model File<br/>.joblib]
    API2 --> MODEL
    PROM[Prometheus<br/>Metrics] --> API1
    PROM --> API2
    GRAF[Grafana<br/>Dashboard] --> PROM

▶ Example: A Complete Production Deployment Architecture

YAML

# docker-compose.yml
version: "3.8"

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - REDIS_URL=redis://redis:6379
    depends_on:
      - redis
    deploy:
      replicas: 2
    restart: unless-stopped

  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/conf.d/default.conf
    depends_on:
      - api
    restart: unless-stopped

volumes:
  redis_data:

TEXT

# nginx.conf (simplified load balancer)
upstream api_servers {
    server api:8000;
}

server {
    listen 80;
    location / {
        proxy_pass http://api_servers;
        proxy_set_header Host $host;
    }
}

▶ Example: API + Redis Cache

PYTHON

# Enhanced app.py with Redis caching
from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np
import hashlib
import json

app = FastAPI(title="SalesPredict API with Cache")
model = joblib.load("salespredict_model.joblib")

# Redis cache (conceptual)
# import redis
# redis_client = redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379"))

class PredictionInput(BaseModel):
    ad_spend_k: float
    traffic_k: float
    category_electronics: float = 0
    category_clothing: float = 0
    is_promotion: float = 0

def get_cache_key(input_data: PredictionInput) -> str:
    data_str = json.dumps(input_data.model_dump(), sort_keys=True)
    return f"pred:{hashlib.md5(data_str.encode()).hexdigest()}"

@app.post("/predict")
def predict(input_data: PredictionInput):
    cache_key = get_cache_key(input_data)

    # Check cache first
    # cached = redis_client.get(cache_key)
    # if cached:
    #     return json.loads(cached)

    features = np.array([[input_data.ad_spend_k, input_data.traffic_k,
                           input_data.category_electronics,
                           input_data.category_clothing, input_data.is_promotion]])
    prediction = float(model.predict(features)[0])
    result = {"predicted_revenue_k": round(prediction, 2)}

    # Cache for 5 minutes
    # redis_client.setex(cache_key, 300, json.dumps(result))

    return result

Output:

TEXT

# Function defined successfully

Component	Role	Technology Choice
API service	Model inference	FastAPI + Uvicorn
Cache	Caching hot predictions	Redis (TTL 5min)
Load balancer	Request distribution	Nginx
Container orchestration	Service management	docker-compose
Health checks	Failure detection	/health + HEALTHCHECK

❓ FAQ

Q Which is better, pickle or joblib?

A Use joblib for sklearn models (it compresses large numpy arrays more efficiently); use pickle for general Python objects. Both carry security risks (an untrusted pickle file can execute malicious code), so MLflow or ONNX is safer for production.

Q Should I use FastAPI or Flask?

A Use FastAPI for new projects — automatic docs (Swagger), type validation (Pydantic), async support, and better performance. Flask is more mature, but its API development experience doesn't match FastAPI.

Q What if my Docker image is too large?

A Three tricks — 1) multi-stage builds (the build stage doesn't end up in the final image); 2) use slim/alpine base images; 3) use .dockerignore to exclude .git/data and the like.

Q How do I update the model without downtime?

A Two approaches — 1) blue-green deployment (switching between old and new versions); 2) rolling updates (docker-compose rolling update). Pair this with MLflow Model Registry to manage versions.

Q How do I optimize API latency?

A Four layers of optimization — 1) cache hot requests with Redis; 2) batch predictions to reduce overhead; 3) run multiple workers in parallel (Uvicorn workers); 4) quantize the model (shrink its size).

Q How do I rate-limit API requests?

A Use the slowapi library for rate limiting — limiter = Limiter(key_func=get_remote_address), e.g. capping requests at 100 per minute. This prevents abuse and overload.

📖 Summary

Model serialization: use joblib for sklearn, torch.save (state_dict) for PyTorch, and MLflow is recommended for production
FastAPI provides the REST API: Pydantic validates inputs, Swagger auto-generates docs, and async handles high concurrency
Docker containerization: multi-stage builds shrink the image, a non-root user hardens security, and HEALTHCHECK monitors health
docker-compose orchestration: a production-grade deployment with API + Redis cache + Nginx load balancing
Caching strategy: Redis caches hot prediction results with a 5-minute TTL, achieving a 30-50% hit rate
API latency target: <50ms (including model inference), with batch predictions further lowering the amortized latency

📝 Exercises

Basic (difficulty ⭐): Train an sklearn model, save it with joblib, then load it in a separate Python script and make a prediction. Hint: joblib.dump/load.
Intermediate (difficulty ⭐⭐): Create a /predict endpoint with FastAPI, including Pydantic input validation and a /health check, then run it with uvicorn and test it with curl. Hint: refer to the complete API code in Section 4.
Challenge (difficulty ⭐⭐⭐): Write a Dockerfile (multi-stage build) + docker-compose.yml (API + Redis + Nginx), build the image and run the full service stack, then verify load balancing and caching. Hint: refer to the config files in Sections 5-6.

← Previous: Intro to MLOps | Next: A/B Testing →

Web-Tutorial Tech Team

A team of developers maintaining programming tutorials. Each tutorial is written and reviewed by developers with expertise in that field. We work to keep our content accurate and reliable — if you spot an issue, please let us know.

About Us

Model Deployment with FastAPI and Docker

1. What You'll Learn

2. A Real ML Engineer's Story

(1) The Pain: A Notebook Model Can't Serve the Business

(2) The FastAPI + Docker Solution

(3) The Payoff: Millions of Requests a Day After Launch

3. Model Serialization

(1) Saving and Loading Models

▶ Example: sklearn Model Serialization

▶ Example: Saving a PyTorch Model

4. FastAPI Model Serving

(1) FastAPI Basics

▶ Example: A Complete Prediction API

(2) Running the FastAPI Service

5. Docker Containerization

(1) Writing a Dockerfile

▶ Example: The SalesPredict Docker Image

(2) Docker Commands

6. docker-compose Orchestration

▶ Example: A Complete Production Deployment Architecture

▶ Example: API + Redis Cache

❓ FAQ

📖 Summary

📝 Exercises

🙏 帮我们做得更好