Most data science tutorials end right where the hard part begins. You train a model, you get great accuracy, and then… it just sits in a Jupyter notebook. Nobody can use it. It runs on your laptop. It breaks the moment someone else tries to run it on their machine.
That’s exactly the problem this tutorial solves.
I’m going to walk you through taking a trained machine learning model and turning it into a real, working REST API that anyone can call over the internet, and then packaging that API inside a Docker container so it runs consistently anywhere: your laptop, a Linux server, AWS, Azure, or Google Cloud.
By the end of this guide, you’ll have:
- A trained ML model saved to disk
- A FastAPI REST API that serves live predictions
- A Dockerized container that packages everything together
- A health check endpoint and proper error handling
- A tested, working deployment you can send to a server
No prior Docker experience needed. I’ll explain every line.
Read: Machine Learning Prod uct Manager
What You’ll Build
We’re building a customer churn prediction API for a fictional SaaS company called ClearPath Analytics based in Denver, Colorado. The API will accept customer behavior data and return a prediction: will this customer churn in the next 30 days?
Here’s what the final project structure looks like:
churn-api/
│
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI application
│ ├── model.py # Model loading logic
│ └── schemas.py # Pydantic input/output models
│
├── model/
│ ├── churn_model.pkl # Trained model
│ └── scaler.pkl # Feature scaler
│
├── train_model.py # Script to train and save the model
├── Dockerfile # Docker build instructions
├── requirements.txt # Python dependencies
├── .dockerignore # Files to exclude from Docker image
└── test_api.py # API tests
Prerequisites
Before we start, make sure you have these installed:
- Python 3.10+ —
python --version - Docker Desktop — Download from docker.com
- pip — comes with Python
Install the required Python libraries:
pip install fastapi uvicorn scikit-learn pandas numpy joblib pydantic
Step 1: Train and Save the Model
First, we need a trained model. In a real project, this would be your existing model pipeline. For this tutorial, I’ll train a LightGBM classifier on synthetic customer data that mirrors what a SaaS company like ClearPath Analytics would have.
Create a file called train_model.py:
# train_model.py
import pandas as pd
import numpy as np
import joblib
import os
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import classification_report
# Create the model/ directory if it doesn't exist
os.makedirs("model", exist_ok=True)
# -----------------------------------------------
# Generate synthetic ClearPath Analytics customer data
# In a real project, you'd load from your database or CSV
# -----------------------------------------------
np.random.seed(42)
n_samples = 2000
data = pd.DataFrame({
'monthly_active_days': np.random.randint(1, 30, n_samples),
'num_features_used': np.random.randint(1, 20, n_samples),
'support_tickets_opened': np.random.poisson(1.5, n_samples),
'days_since_last_login': np.random.exponential(15, n_samples).astype(int),
'contract_length_months': np.random.choice([1, 6, 12, 24], n_samples),
'monthly_spend_usd': np.random.uniform(49, 999, n_samples).round(2),
'num_team_members': np.random.randint(1, 50, n_samples),
'billing_failures': np.random.poisson(0.3, n_samples),
'onboarding_completed': np.random.choice([0, 1], n_samples, p=[0.2, 0.8]),
'nps_score': np.random.randint(0, 11, n_samples),
})
# Build a realistic churn label
# Customers with low activity, high tickets, and low NPS are more likely to churn
churn_score = (
-0.4 * data['monthly_active_days'] +
0.5 * data['support_tickets_opened'] +
0.3 * data['days_since_last_login'] +
-0.3 * data['nps_score'] +
0.4 * data['billing_failures'] +
-0.2 * data['onboarding_completed']
)
churn_prob = 1 / (1 + np.exp(-0.1 * churn_score))
data['churned'] = (churn_prob > 0.5).astype(int)
print(f"Dataset shape: {data.shape}")
print(f"Churn rate: {data['churned'].mean():.2%}")
# -----------------------------------------------
# Train/test split
# -----------------------------------------------
X = data.drop('churned', axis=1)
y = data['churned']
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42, stratify=y
)
# -----------------------------------------------
# Scale features — fit ONLY on training data
# -----------------------------------------------
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)
# -----------------------------------------------
# Train model
# -----------------------------------------------
model = GradientBoostingClassifier(
n_estimators=100,
learning_rate=0.1,
max_depth=4,
random_state=42
)
model.fit(X_train_scaled, y_train)
# -----------------------------------------------
# Evaluate
# -----------------------------------------------
y_pred = model.predict(X_test_scaled)
print("\n=== Model Evaluation ===")
print(classification_report(y_test, y_pred, target_names=['Retained', 'Churned']))
# -----------------------------------------------
# Save model and scaler
# -----------------------------------------------
joblib.dump(model, 'model/churn_model.pkl')
joblib.dump(scaler, 'model/scaler.pkl')
joblib.dump(list(X.columns), 'model/feature_names.pkl')
print("\n✅ Model saved to model/churn_model.pkl")
print("✅ Scaler saved to model/scaler.pkl")
print("✅ Feature names saved to model/feature_names.pkl")
Run this:
python train_model.py
Output:
Dataset shape: (2000, 11)
Churn rate: 37.20%
=== Model Evaluation ===
precision recall f1-score support
Retained 0.88 0.92 0.90 251
Churned 0.85 0.78 0.81 149
accuracy 0.87 400
✅ Model saved to model/churn_model.pkl
✅ Scaler saved to model/scaler.pkl
✅ Feature names saved to model/feature_names.pkl
You can see the output in the screenshot below.
You now have three .pkl files in your model/ folder. These are what the API will load at startup.
Step 2: Define the Request and Response Schemas
This is a step most deployment tutorials rush. The schema is what defines what data your API accepts and what it returns. Getting this right prevents a huge class of bugs.
Create app/schemas.py:
# app/schemas.py
from pydantic import BaseModel, Field, field_validator
from typing import Literal
class CustomerFeatures(BaseModel):
"""
Input schema for churn prediction.
All fields validated automatically by Pydantic.
"""
monthly_active_days: int = Field(
..., ge=0, le=31,
description="Number of days the customer was active in the last month"
)
num_features_used: int = Field(
..., ge=0, le=100,
description="Number of product features used in the last 30 days"
)
support_tickets_opened: int = Field(
..., ge=0,
description="Support tickets opened in the last 90 days"
)
days_since_last_login: int = Field(
..., ge=0,
description="Days since the customer last logged in"
)
contract_length_months: int = Field(
..., ge=1,
description="Contract length in months (1, 6, 12, or 24)"
)
monthly_spend_usd: float = Field(
..., gt=0,
description="Monthly subscription spend in USD"
)
num_team_members: int = Field(
..., ge=1,
description="Number of team members on the account"
)
billing_failures: int = Field(
..., ge=0,
description="Number of billing failures in the last 6 months"
)
onboarding_completed: Literal[0, 1] = Field(
...,
description="Whether the customer completed onboarding (1=yes, 0=no)"
)
nps_score: int = Field(
..., ge=0, le=10,
description="Net Promoter Score from last survey (0-10)"
)
class Config:
json_schema_extra = {
"example": {
"monthly_active_days": 8,
"num_features_used": 3,
"support_tickets_opened": 4,
"days_since_last_login": 22,
"contract_length_months": 1,
"monthly_spend_usd": 99.00,
"num_team_members": 2,
"billing_failures": 1,
"onboarding_completed": 0,
"nps_score": 3
}
}
class PredictionResponse(BaseModel):
"""Output schema for churn prediction response."""
prediction: int = Field(..., description="0 = Retained, 1 = Churned")
label: str = Field(..., description="Human-readable prediction label")
churn_probability: float = Field(..., description="Probability of churn (0.0 to 1.0)")
risk_level: str = Field(..., description="Risk category: Low, Medium, or High")
model_version: str = Field(..., description="Model version used for prediction")
class HealthResponse(BaseModel):
"""Health check response schema."""
status: str
model_loaded: bool
model_version: str
The Field(…, ge=0, le=31) syntax means “required, must be >= 0 and <= 31.” FastAPI uses Pydantic to validate every incoming request automatically — before your model ever sees the data. Invalid input returns a clean 422 error instead of a cryptic Python traceback.
Step 3: Build the Model Loading Logic
Create app/model.py:
# app/model.py
import joblib
import numpy as np
import pandas as pd
from pathlib import Path
MODEL_VERSION = "1.0.0"
class ChurnPredictor:
"""
Handles model loading and prediction logic.
Loaded once at API startup — not on every request.
"""
def __init__(self):
self.model = None
self.scaler = None
self.feature_names = None
self.is_loaded = False
def load(self):
"""Load model artifacts from disk."""
model_dir = Path("model")
if not model_dir.exists():
raise FileNotFoundError(
"model/ directory not found. Run train_model.py first."
)
self.model = joblib.load(model_dir / "churn_model.pkl")
self.scaler = joblib.load(model_dir / "scaler.pkl")
self.feature_names = joblib.load(model_dir / "feature_names.pkl")
self.is_loaded = True
print(f"✅ Model v{MODEL_VERSION} loaded successfully")
def predict(self, features: dict) -> dict:
"""
Run inference on a single customer record.
Returns prediction, probability, and risk level.
"""
if not self.is_loaded:
raise RuntimeError("Model not loaded. Call load() first.")
# Build DataFrame with correct feature order
input_df = pd.DataFrame([features])[self.feature_names]
# Scale using the same scaler fitted on training data
input_scaled = self.scaler.transform(input_df)
# Predict
prediction = int(self.model.predict(input_scaled)[0])
probabilities = self.model.predict_proba(input_scaled)[0]
churn_prob = round(float(probabilities[1]), 4)
# Assign risk level
if churn_prob < 0.3:
risk_level = "Low"
elif churn_prob < 0.6:
risk_level = "Medium"
else:
risk_level = "High"
return {
"prediction": prediction,
"label": "Churned" if prediction == 1 else "Retained",
"churn_probability": churn_prob,
"risk_level": risk_level,
"model_version": MODEL_VERSION
}
# Single global instance — loaded once when the API starts
predictor = ChurnPredictor()
Loading the model once at startup (not per request) is critical for performance. If you load it on every request, your API will be 10-20x slower than it needs to be.
Check out: Machine Learning Engineering with Python
Step 4: Build the FastAPI Application
This is the core of everything. Create app/main.py:
# app/main.py
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from contextlib import asynccontextmanager
import logging
import time
from app.schemas import CustomerFeatures, PredictionResponse, HealthResponse
from app.model import predictor, MODEL_VERSION
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# -----------------------------------------------
# Lifespan: runs on startup and shutdown
# This is the modern FastAPI way (replaces @app.on_event)
# -----------------------------------------------
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: load model
logger.info("Starting ClearPath Churn Prediction API...")
predictor.load()
logger.info("API ready to serve predictions.")
yield
# Shutdown: cleanup if needed
logger.info("Shutting down API...")
# -----------------------------------------------
# Create FastAPI app
# -----------------------------------------------
app = FastAPI(
title="ClearPath Churn Prediction API",
description=(
"Predicts the likelihood of a customer churning in the next 30 days. "
"Built for ClearPath Analytics — Denver, CO."
)