A machine learning model locked in a Jupyter notebook delivers zero business value. Wrapping it in an API turns it into a service that any application, website, or system can consume. FastAPI is the modern choice for ML APIs — fast, auto-documented, and type-safe. This guide builds production-ready ML APIs from scratch.
Why FastAPI Over Flask?
Flask is simpler and widely used, but FastAPI has decisive advantages for ML APIs: automatic input validation with Pydantic, interactive documentation at /docs out of the box, async support for non-blocking I/O, and performance comparable to Node.js. Flask is fine for simple APIs with few endpoints; choose FastAPI for anything you plan to maintain or scale.
Your First FastAPI ML API
pip install fastapi uvicorn pydantic scikit-learn joblib
# train.py — train and save model
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import load_iris
import joblib
X, y = load_iris(return_X_y=True)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
joblib.dump(model, 'model.pkl')
print('Model saved')
# main.py — FastAPI application
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, validator
from typing import List
import joblib
import numpy as np
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title='Iris Classifier API',
description='Predict iris species from measurements',
version='1.0.0')
model = joblib.load('model.pkl')
CLASSES = ['setosa', 'versicolor', 'virginica']
class PredictRequest(BaseModel):
sepal_length: float = Field(..., gt=0, le=20, example=5.1)
sepal_width: float = Field(..., gt=0, le=20, example=3.5)
petal_length: float = Field(..., gt=0, le=20, example=1.4)
petal_width: float = Field(..., gt=0, le=20, example=0.2)
class PredictResponse(BaseModel):
prediction: str
class_index: int
probabilities: dict[str, float]
confidence: float
class BatchRequest(BaseModel):
instances: List[PredictRequest]
@app.get('/health')
async def health():
return {'status': 'ok', 'model': 'RandomForestClassifier'}
@app.post('/predict', response_model=PredictResponse)
async def predict(request: PredictRequest):
features = np.array([[
request.sepal_length, request.sepal_width,
request.petal_length, request.petal_width
]])
pred = int(model.predict(features)[0])
probs = model.predict_proba(features)[0]
logger.info(f'Prediction: {CLASSES[pred]} ({probs[pred]:.4f})')
return PredictResponse(
prediction=CLASSES[pred],
class_index=pred,
probabilities={cls: round(float(p), 4)
for cls, p in zip(CLASSES, probs)},
confidence=round(float(probs[pred]), 4)
)
@app.post('/predict/batch')
async def predict_batch(request: BatchRequest):
features = np.array([[
r.sepal_length, r.sepal_width,
r.petal_length, r.petal_width
] for r in request.instances])
preds = model.predict(features)
probs = model.predict_proba(features)
return {'predictions': [
{'prediction': CLASSES[p], 'confidence': round(float(pr[p]), 4)}
for p, pr in zip(preds, probs)
]}
# Run: uvicorn main:app --reload --port 8000
# Docs: http://localhost:8000/docs
Adding Authentication
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import secrets
security = HTTPBearer()
API_KEYS = {'sk-prod-abc123', 'sk-dev-xyz789'} # store in env vars in production
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
token = credentials.credentials
if token not in API_KEYS:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail='Invalid API key'
)
return token
# Protect endpoints with Depends
@app.post('/predict/secure', response_model=PredictResponse)
async def predict_secure(request: PredictRequest,
token: str = Depends(verify_token)):
return await predict(request)
Async Processing for Heavy Models
import asyncio
from fastapi import BackgroundTasks
import uuid
# In-memory job store (use Redis in production)
jobs = {}
@app.post('/predict/async')
async def predict_async(request: PredictRequest,
background_tasks: BackgroundTasks):
job_id = str(uuid.uuid4())
jobs[job_id] = {'status': 'processing'}
async def process():
await asyncio.sleep(0) # yield to event loop
features = np.array([[
request.sepal_length, request.sepal_width,
request.petal_length, request.petal_width
]])
pred = int(model.predict(features)[0])
jobs[job_id] = {
'status': 'complete',
'prediction': CLASSES[pred]
}
background_tasks.add_task(process)
return {'job_id': job_id, 'status': 'processing'}
@app.get('/jobs/{job_id}')
async def get_job(job_id: str):
if job_id not in jobs:
raise HTTPException(status_code=404, detail='Job not found')
return jobs[job_id]
Error Handling and Logging
from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
logger.error(f'ValueError: {exc}')
return JSONResponse(
status_code=422,
content={'error': 'Validation error', 'detail': str(exc)}
)
@app.middleware('http')
async def log_requests(request: Request, call_next):
import time
start = time.time()
response = await call_next(request)
duration = time.time() - start
logger.info(f'{request.method} {request.url.path} '
f'{response.status_code} {duration:.3f}s')
return response
Dockerising the API
# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
# requirements.txt
fastapi==0.111.0
uvicorn[standard]==0.30.0
pydantic==2.7.0
scikit-learn==1.5.0
joblib==1.4.0
numpy==1.26.4
docker build -t ml-api:v1 .
docker run -p 8000:8000 ml-api:v1
# Test
curl -X POST http://localhost:8000/predict -H 'Content-Type: application/json' -d '{"sepal_length":5.1,"sepal_width":3.5,"petal_length":1.4,"petal_width":0.2}'
Conclusion
FastAPI transforms your ML model from a notebook experiment into a production service in under 100 lines of code. The automatic /docs page eliminates the need to write API documentation. Pydantic validation catches bad inputs before they crash your model. Adding authentication and Docker packaging takes another hour and makes your API production-ready. Every data scientist who can build, document, and deploy an ML API is dramatically more valuable than one who can only build models.


