Building APIs with FastAPI: From Endpoint to Deployment
/docs page generated for free.After this lesson, you will be able to:
- Explain what an API is, how REST works, and why HTTP status codes matter
- Build a FastAPI application with path parameters, query parameters, and request bodies
- Use Pydantic models for request validation, serialization, and nested data structures
- Serve a trained ML model behind a /predict endpoint with batch prediction support
- Add error handling, CORS middleware, and basic authentication to a FastAPI app
- Write integration tests for API endpoints using httpx and TestClient
- Distinguish ASGI from WSGI and understand why FastAPI is async-first
Before You Start
#Why APIs? REST, HTTP, and JSON
#HTTP Methods
| Method | Purpose | Example | Analogy |
|---|---|---|---|
GET | Read data | Get a user's profile | Looking up a book in a library |
POST | Create data | Submit a new prediction request | Filling out a form |
PUT | Replace data entirely | Update all user fields | Rewriting an entire chapter |
PATCH | Update data partially | Change just the user's email | Editing one paragraph |
DELETE | Remove data | Delete a saved model | Throwing away a file |
#HTTP Status Codes
Status codes tell the client what happened:
200 OK -- Request succeeded, here is the data
201 Created -- A new resource was created
204 No Content -- Success, but nothing to return (e.g., after DELETE)
400 Bad Request -- Client sent invalid data
401 Unauthorized -- Client is not authenticated
403 Forbidden -- Client is authenticated but not allowed
404 Not Found -- The resource does not exist
422 Unprocessable -- Data is syntactically correct but semantically invalid
500 Internal Error -- Something broke on the server
#JSON — The Universal Data Format
APIs exchange data as JSON (JavaScript Object Notation). It maps directly to Python dictionaries:
# JSON
{
"name": "Alice",
"age": 28,
"scores": [95, 87, 92],
"address": {
"city": "San Francisco",
"state": "CA"
}
}
# Python equivalent
data = {
"name": "Alice",
"age": 28,
"scores": [95, 87, 92],
"address": {
"city": "San Francisco",
"state": "CA"
}
}#FastAPI Basics — Your First API
#Installation
pip install fastapi uvicorn
fastapi is the framework. uvicorn is the ASGI server that runs it.#Hello World
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def root():
"""Root endpoint -- returns a welcome message."""
return {"message": "Hello, World!"}
@app.get("/health")
def health_check():
"""Health check endpoint for monitoring."""
return {"status": "healthy"}Run it:
uvicorn main:app --reload
main= the Python file (main.py)app= the FastAPI instance--reload= auto-restart on code changes (development only)
http://localhost:8000 and you see {"message": "Hello, World!"}. Visit http://localhost:8000/docs and you get automatic interactive documentation powered by Swagger UI — this is one of FastAPI's killer features.#Path Parameters
Path parameters are variables embedded in the URL:
@app.get("/users/{user_id}")
def get_user(user_id: int):
"""Get a user by their ID."""
return {"user_id": user_id, "name": f"User #{user_id}"}
@app.get("/models/{model_name}/version/{version}")
def get_model_version(model_name: str, version: int):
"""Get a specific version of an ML model."""
return {
"model": model_name,
"version": version,
"status": "active"
}version is annotated int. A client requests /models/bert/version/latest. What does FastAPI return?
/users/abc, you get a 422 error because abc is not an integer. No manual validation needed.#Query Parameters
? in a URL:from typing import Optional
@app.get("/predictions")
def list_predictions(
model_name: str,
limit: int = 10,
offset: int = 0,
status: Optional[str] = None,
):
"""List predictions with pagination and optional filtering."""
return {
"model": model_name,
"limit": limit,
"offset": offset,
"status_filter": status,
}GET /predictions?model_name=classifier&limit=5&status=completedFastAPI distinguishes path params from query params automatically: if the parameter is in the path, it is a path param. Otherwise, it is a query param.
#Request Body with Pydantic
POST and PUT requests, you send data in the request body. FastAPI uses Pydantic models to define and validate the shape of that data:from pydantic import BaseModel
class PredictionRequest(BaseModel):
"""Schema for a prediction request."""
features: list[float]
model_name: str = "default"
return_probabilities: bool = False
@app.post("/predict")
def predict(request: PredictionRequest):
"""Make a prediction based on input features."""
# request.features is already validated as list[float]
# request.model_name defaults to "default" if not provided
prediction = sum(request.features) / len(request.features) # placeholder
return {
"prediction": prediction,
"model_used": request.model_name,
"input_dim": len(request.features),
}Send a POST request with JSON body:
{
"features": [1.5, 2.3, 0.7, 4.1],
"model_name": "linear_v2",
"return_probabilities": true
}
FastAPI automatically parses the JSON, validates it against the Pydantic model, and returns a 422 error with detailed messages if validation fails.
You send a POST request to /predict with body {"features": "not a list", "model_name": "test"}. What happens?
#Pydantic Models — Validation, Serialization, and Nested Models
Pydantic is not just for type checking. It handles validation, serialization, default values, and complex nested structures.
#Field Validators
from pydantic import BaseModel, Field, field_validator
class TrainingConfig(BaseModel):
"""Configuration for model training."""
model_name: str = Field(..., min_length=1, max_length=100)
learning_rate: float = Field(default=0.001, gt=0, lt=1)
epochs: int = Field(default=10, ge=1, le=1000)
batch_size: int = Field(default=32, description="Must be a power of 2")
dropout: float = Field(default=0.1, ge=0, le=1)
@field_validator("batch_size")
@classmethod
def batch_size_must_be_power_of_two(cls, v):
if v & (v - 1) != 0:
raise ValueError(f"batch_size must be a power of 2, got {v}")
return v
@field_validator("model_name")
@classmethod
def model_name_must_be_lowercase(cls, v):
if v != v.lower():
raise ValueError("model_name must be lowercase")
return vField(...) means "this field is required" (no default). Field(default=0.001, gt=0) means "defaults to 0.001 and must be greater than 0."#Nested Models
class DatasetInfo(BaseModel):
name: str
num_samples: int = Field(ge=1)
num_features: int = Field(ge=1)
target_column: str
class ModelMetrics(BaseModel):
accuracy: float = Field(ge=0, le=1)
precision: float = Field(ge=0, le=1)
recall: float = Field(ge=0, le=1)
f1_score: float = Field(ge=0, le=1)
class TrainingResult(BaseModel):
"""Complete training result with nested models."""
config: TrainingConfig
dataset: DatasetInfo
metrics: ModelMetrics
training_time_seconds: float
model_path: str
# FastAPI automatically handles nested validation
@app.post("/train", response_model=TrainingResult)
def train_model(config: TrainingConfig, dataset: DatasetInfo):
"""Train a model and return the results."""
# ... training logic ...
return TrainingResult(
config=config,
dataset=dataset,
metrics=ModelMetrics(
accuracy=0.92, precision=0.89,
recall=0.91, f1_score=0.90,
),
training_time_seconds=142.5,
model_path="/models/classifier_v3.pkl",
)#Response Models
response_model parameter in the decorator controls what gets sent back to the client. It filters out any extra fields and validates the response:class UserPublic(BaseModel):
"""Public user info -- excludes sensitive fields."""
id: int
name: str
email: str
class UserInternal(BaseModel):
"""Internal user info -- includes everything."""
id: int
name: str
email: str
hashed_password: str
api_key: str
@app.get("/users/{user_id}", response_model=UserPublic)
def get_user(user_id: int):
# Even though we return all fields internally,
# FastAPI strips hashed_password and api_key from the response
user = get_user_from_db(user_id) # returns UserInternal
return user`get_user` returns a `UserInternal` object containing `hashed_password`, yet the client never sees that field. What removes it?
#ML Model Serving — The /predict Endpoint
This is the most common use case for FastAPI in the ML world: serving a trained model behind an API.
#Loading a Trained Model
import pickle
from pathlib import Path
from fastapi import FastAPI
from pydantic import BaseModel, Field
import numpy as np
app = FastAPI(title="ML Prediction Service", version="1.0.0")
# Load model at startup (once, not per request)
MODEL_PATH = Path("models/classifier.pkl")
def load_model():
"""Load the trained model from disk."""
with open(MODEL_PATH, "rb") as f:
return pickle.load(f)
# Use a lifespan event to load the model once at startup
from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Load ML model on startup, clean up on shutdown."""
app.state.model = load_model()
print(f"Model loaded from {MODEL_PATH}")
yield
print("Shutting down, cleaning up...")
app = FastAPI(title="ML Prediction Service", lifespan=lifespan)#Single Prediction Endpoint
class PredictRequest(BaseModel):
features: list[float] = Field(..., min_length=1)
class PredictResponse(BaseModel):
prediction: int
confidence: float = Field(ge=0, le=1)
model_version: str
@app.post("/predict", response_model=PredictResponse)
def predict_single(request: PredictRequest):
"""Make a single prediction."""
model = app.state.model
features = np.array(request.features).reshape(1, -1)
prediction = model.predict(features)[0]
probabilities = model.predict_proba(features)[0]
confidence = float(max(probabilities))
return PredictResponse(
prediction=int(prediction),
confidence=confidence,
model_version="1.0.0",
)#Batch Prediction Endpoint
class BatchPredictRequest(BaseModel):
samples: list[list[float]] = Field(..., min_length=1, max_length=1000)
class BatchPredictResponse(BaseModel):
predictions: list[int]
confidences: list[float]
count: int
model_version: str
@app.post("/predict/batch", response_model=BatchPredictResponse)
def predict_batch(request: BatchPredictRequest):
"""Make predictions on a batch of samples (up to 1000)."""
model = app.state.model
features = np.array(request.samples)
predictions = model.predict(features)
probabilities = model.predict_proba(features)
confidences = [float(max(row)) for row in probabilities]
return BatchPredictResponse(
predictions=[int(p) for p in predictions],
confidences=confidences,
count=len(predictions),
model_version="1.0.0",
)#Model Metadata Endpoint
@app.get("/model/info")
def model_info():
"""Return metadata about the currently loaded model."""
model = app.state.model
return {
"model_type": type(model).__name__,
"n_features": model.n_features_in_,
"classes": model.classes_.tolist(),
"model_version": "1.0.0",
}You load a large ML model inside the /predict endpoint function (not at startup). What happens when 100 users hit the endpoint simultaneously?
#Error Handling, Middleware, CORS, and Authentication
#Custom Error Handling
from fastapi import FastAPI, HTTPException, status
@app.post("/predict")
def predict(request: PredictRequest):
"""Predict with proper error handling."""
model = app.state.model
# Check feature dimensions
expected_features = model.n_features_in_
if len(request.features) != expected_features:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Expected {expected_features} features, got {len(request.features)}",
)
try:
features = np.array(request.features).reshape(1, -1)
prediction = model.predict(features)[0]
return {"prediction": int(prediction)}
except Exception as e:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=f"Prediction failed: {str(e)}",
)#CORS Middleware
localhost:3000 tries to call your API on localhost:8000, the browser blocks it by default (Cross-Origin Resource Sharing policy). You need CORS middleware:from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:3000", "https://myapp.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)#Request Logging Middleware
import time
import logging
logger = logging.getLogger("api")
@app.middleware("http")
async def log_requests(request, call_next):
"""Log every request with timing information."""
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#Basic API Key Authentication
from fastapi import Depends, Security
from fastapi.security import APIKeyHeader
import os
API_KEY = os.environ.get("API_KEY", "dev-key-change-me")
api_key_header = APIKeyHeader(name="X-API-Key")
def verify_api_key(api_key: str = Security(api_key_header)):
"""Verify the API key from the request header."""
if api_key != API_KEY:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid API key",
)
return api_key
@app.post("/predict", dependencies=[Depends(verify_api_key)])
def predict_secure(request: PredictRequest):
"""Secured prediction endpoint -- requires valid API key."""
# ... prediction logic ...
return {"prediction": 1}The auth dependency is attached as `dependencies=[Depends(verify_api_key)]` on the decorator, not as a parameter of `predict_secure`. What does that change?
Dependency Injection with Depends
Depends system lets you declare reusable logic (authentication, database sessions, rate limiting) that runs before your endpoint and injects its result as a parameter. This is called dependency injection.from fastapi import Depends, HTTPException, status
from sqlalchemy.orm import Session
# --- Database session dependency ---
def get_db():
"""Provide a database session for the duration of one request."""
db = SessionLocal() # create a new session
try:
yield db # inject the session into the endpoint
db.commit() # commit if no exception
except Exception:
db.rollback()
raise
finally:
db.close() # always close, even if an exception occurred
@app.get("/experiments/{exp_id}")
def get_experiment(
exp_id: int,
db: Session = Depends(get_db), # FastAPI calls get_db() and injects the result
):
"""Fetch an experiment by ID. db is automatically provided."""
exp = db.query(Experiment).filter(Experiment.id == exp_id).first()
if not exp:
raise HTTPException(status_code=404, detail=f"Experiment {exp_id} not found")
return exp
# --- Composable dependencies: auth + db ---
def get_current_user(api_key: str = Security(api_key_header)) -> str:
if api_key != VALID_API_KEY:
raise HTTPException(status_code=401, detail="Invalid API key")
return api_key # or return the user object from DB
@app.post("/predict", dependencies=[Depends(get_current_user)])
def predict_secured(
request: PredictRequest,
db: Session = Depends(get_db),
):
"""Endpoint protected by both authentication AND a database session."""
# FastAPI resolved get_current_user (auth check) and get_db (DB session)
# before this function body runs
...Depends works for:- Database sessions. One session per request, committed/rolled back automatically
- Authentication. Verify JWT (JSON Web Token) credentials or API keys, raise 401 if invalid
- Rate limiting. Check request counts against Redis before the endpoint runs
- Settings injection. Inject typed settings without global variables
- Pagination parameters. Standardize
limit/offsetparsing across all list endpoints
An endpoint declares two dependencies that both call Depends(get_db). How many database sessions are created for one request?
#Testing with httpx and TestClient
TestClient that lets you test endpoints without starting a real server:# test_main.py
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_root():
"""Test the root endpoint returns a welcome message."""
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"message": "Hello, World!"}
def test_health_check():
"""Test the health endpoint."""
response = client.get("/health")
assert response.status_code == 200
assert response.json()["status"] == "healthy"
def test_predict_valid_input():
"""Test prediction with valid features."""
response = client.post(
"/predict",
json={"features": [1.0, 2.0, 3.0, 4.0]},
)
assert response.status_code == 200
data = response.json()
assert "prediction" in data
assert "confidence" in data
def test_predict_invalid_features():
"""Test prediction with invalid feature type."""
response = client.post(
"/predict",
json={"features": "not a list"},
)
assert response.status_code == 422
def test_predict_empty_features():
"""Test prediction with empty feature list."""
response = client.post(
"/predict",
json={"features": []},
)
assert response.status_code == 422
def test_predict_wrong_dimensions():
"""Test prediction with wrong number of features."""
response = client.post(
"/predict",
json={"features": [1.0]}, # model expects 4 features
)
assert response.status_code == 400
assert "Expected" in response.json()["detail"]#Testing with pytest Fixtures
import pytest
from fastapi.testclient import TestClient
@pytest.fixture
def client():
"""Create a test client with a fresh app instance."""
from main import app
return TestClient(app)
@pytest.fixture
def auth_headers():
"""Provide authentication headers for secured endpoints."""
return {"X-API-Key": "dev-key-change-me"}
def test_secured_endpoint_without_key(client):
"""Requests without API key should be rejected."""
response = client.post("/predict", json={"features": [1.0, 2.0, 3.0, 4.0]})
assert response.status_code == 401 # or 403
def test_secured_endpoint_with_key(client, auth_headers):
"""Requests with valid API key should succeed."""
response = client.post(
"/predict",
json={"features": [1.0, 2.0, 3.0, 4.0]},
headers=auth_headers,
)
assert response.status_code == 200
@pytest.mark.parametrize("features,expected_status", [
([1.0, 2.0, 3.0, 4.0], 200),
([0.0, 0.0, 0.0, 0.0], 200),
([], 422),
("invalid", 422),
])
def test_predict_various_inputs(client, auth_headers, features, expected_status):
"""Parametrized test for various prediction inputs."""
response = client.post(
"/predict",
json={"features": features},
headers=auth_headers,
)
assert response.status_code == expected_status#Async Testing with httpx
httpx.AsyncClient:import httpx
import pytest
@pytest.mark.asyncio
async def test_predict_async():
"""Test with async client for full async stack testing."""
async with httpx.AsyncClient(app=app, base_url="http://test") as client:
response = await client.post(
"/predict",
json={"features": [1.0, 2.0, 3.0, 4.0]},
)
assert response.status_code == 200#Code Playground
Tests · Build Pydantic models, validate inputs, simulate API endpoints, and handle errors!
#Key Takeaways
- FastAPI validates inputs automatically with Pydantic — define a
BaseModelfor your request body, and FastAPI rejects invalid data with a 422 error before your code runs. No manualif isinstance()checks needed - Load ML models at startup, not per request — use
lifespanevents or global variables to load the model once. Loading per request multiplies memory usage and latency by the number of concurrent users - Use
deffor CPU-bound endpoints,async deffor I/O-bound — putting blocking code (model inference, file I/O) insideasync deffreezes the entire server. FastAPI runs sync functions in a thread pool automatically response_modelcontrols what the client sees — use it to strip sensitive fields (passwords, API keys) from responses and to generate accurate API documentation- TestClient makes API testing trivial — no running server needed. Combine with pytest fixtures and parametrize for comprehensive endpoint testing
- Visit /docs for free interactive documentation — FastAPI generates Swagger UI automatically from your type annotations and Pydantic models
What happens when a POST request sends data that does not match the Pydantic model?