If you’ve built a web API in Python before, you’ve probably used Flask. Flask is minimal: you write route functions, decorate them with @app.route, and return whatever you want. FastAPI keeps the same decorator pattern but adds type annotations throughout. Annotate your function parameters and return types, and FastAPI handles input validation, error responses, and API documentation automatically.
For an ML serving API, three patterns in FastAPI do most of the heavy lifting: lifespan for loading artifacts before the first request, Pydantic models for input validation, and a consistent approach to returning 503 when dependencies aren’t ready. The flight disruption prediction API uses all three.
What You Need Coming In
- Comfortable writing Python functions and classes
- Basic understanding of HTTP: requests, responses, status codes
- No prior FastAPI knowledge needed — Flask experience helps but isn’t required
Lifespan: Loading Artifacts Before Requests Arrive
The serving API needs three things loaded before it can handle a request: the ML model bundle (an ONNX file plus calibrator and SHAP explainer), the gold DuckDB (the pre-aggregated reliability data), and a connection to the live aircraft positions cache. All of these take time and can fail.
The wrong place to load them is inside a route handler. If you load the model on the first /api/predict call, that request takes 2-3 seconds while every other request waits. If loading fails, you get a 500 error on the first real user request with no warning.
FastAPI’s lifespan context manager runs setup code before the app starts accepting requests and cleanup code when it shuts down:
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# --- startup: everything in here runs before the first request ---
settings = get_settings()
# 1. Resolve and load the ML model bundle
resolved = ensure_local_artifacts(settings)
try:
state.artifacts = load_bundle(resolved["bundle_dir"])
state.model_loaded = True
except Exception as exc:
log.error("Failed to load model bundle: %s", exc)
state.model_loaded = False # API starts anyway, returns 503 on predict
# 2. Open the gold DuckDB read-only
try:
state.store = GoldStore(resolved["duckdb_path"])
state.gold_loaded = True
except Exception as exc:
log.error("Failed to open gold DuckDB: %s", exc)
state.gold_loaded = False # reliability endpoints return 503
# 3. Wire up live positions (optional, degrades gracefully)
state.live = LivePositions(settings)
yield # <-- the app runs here, serving requests
# --- shutdown: everything after yield runs on shutdown ---
if state.store is not None:
state.store.close()
if state.live is not None:
state.live.close()
app = FastAPI(title="Flight Disruption Serving API", lifespan=lifespan)The yield is the dividing line: everything before it is startup, everything after is shutdown. The app only starts serving requests after yield is reached — which only happens if startup completes without raising an exception.
Notice the try/except around each loading step. A failure to load the model doesn’t crash the startup — it sets state.model_loaded = False and continues. The app starts successfully, and the prediction endpoint returns a 503 with a clear message rather than crashing or silently returning wrong results. This is more useful in practice than an all-or-nothing startup.
The State Singleton
Artifacts loaded during lifespan need to be accessible in route handlers. The cleanest approach is a module-level class instance that holds everything:
class _State:
settings = None
artifacts = None # ML model bundle
store = None # GoldStore (DuckDB wrapper)
live = None # live aircraft positions
model_loaded = False
gold_loaded = False
state = _State()state is a module-level singleton. All route handlers import and read from it. lifespan populates it during startup. This is simpler than FastAPI’s dependency injection for stateful globals — dependency injection is valuable for per-request resources (database sessions, authentication tokens), but a model bundle that loads once and stays loaded doesn’t need injection overhead.
Using a class rather than plain module-level variables means the state is grouped and easy to reason about. state.model_loaded is clearer than MODEL_LOADED floating in module scope, and the class can be replaced with a mock in tests.
Pydantic Models for Input Validation
The prediction endpoint takes five inputs: two airport codes, a carrier code, a date, and a departure hour. Without validation, you’d write:
@app.post("/api/predict")
def api_predict(origin: str, dest: str, carrier: str, date: str, dep_hour: int):
if not origin or len(origin) != 3:
return {"error": "invalid origin"}
# ... more manual checksPydantic models eliminate that. You define a class with annotated fields, and FastAPI validates incoming JSON against it automatically:
from pydantic import BaseModel, Field
class PredictRequest(BaseModel):
origin: str = Field(..., min_length=3, max_length=4, examples=["ATL"])
dest: str = Field(..., min_length=3, max_length=4, examples=["ORD"])
carrier: str = Field(..., min_length=2, max_length=3, examples=["DL"])
date: str = Field(..., examples=["2026-06-20"])
dep_hour: int = Field(..., ge=0, le=23, examples=[17])... as the first argument means the field is required. min_length/max_length validate string length. ge/le validate numeric bounds (greater-than-or-equal, less-than-or-equal). If a request sends dep_hour: 25 or origin: "X", FastAPI returns a 422 Unprocessable Entity response with a clear error message — before your route function even runs.
The examples values appear in the auto-generated OpenAPI documentation at /docs. Anyone hitting the API gets an interactive documentation page with realistic example values they can use immediately.
The route handler then takes the model as a parameter:
@app.post("/api/predict")
def api_predict(req: PredictRequest) -> dict:
_require_model()
_require_gold()
origin = req.origin.upper() # normalize to uppercase
dest = req.dest.upper()
carrier = req.carrier.upper()
try:
return run_predict(
store=state.store,
artifacts=state.artifacts,
origin=origin,
dest=dest,
carrier=carrier,
date_str=req.date,
dep_hour=req.dep_hour,
)
except ValueError as exc:
raise HTTPException(status_code=422, detail=str(exc))FastAPI reads the type annotation req: PredictRequest, parses the request body as JSON, validates it against the Pydantic model, and passes the validated object to the function. If validation fails, the 422 response goes out automatically.
Graceful Degradation with _require_model and _require_gold
Each endpoint checks its dependencies before doing any work:
def _require_model() -> None:
if not state.model_loaded or state.artifacts is None:
raise HTTPException(status_code=503, detail="model not loaded")
def _require_gold() -> None:
if not state.gold_loaded or state.store is None:
raise HTTPException(status_code=503, detail="gold data not loaded")HTTPException with status 503 (Service Unavailable) is the right response when a dependency isn’t ready — it tells the caller “the service is running but this specific capability isn’t available right now.” A 500 (Internal Server Error) implies a bug; a 503 implies a known unavailability.
The reliability and airport endpoints only call _require_gold(). If the model fails to load but DuckDB loaded fine, those endpoints still work — users can see historical reliability data even if predictions are down. The live positions endpoint doesn’t require either, since it reads from a separate cache.
This layered availability is possible because the startup logic loads each dependency independently and sets a flag per dependency rather than one global “ready” flag.
The Health Endpoint
The health endpoint exposes startup state to monitoring systems and load balancers:
@app.get("/health")
def health() -> dict:
return {
"status": "ok",
"model_loaded": state.model_loaded,
"gold_loaded": state.gold_loaded,
"data_version": state.settings.data_version if state.settings else "unknown",
}"status": "ok" always returns as long as the process is alive — a load balancer uses this to confirm the process is running. model_loaded and gold_loaded tell you which capabilities are available. data_version tracks which vintage of training data the model was built from.
The distinction matters in practice. An orchestrator doing a health check to decide whether to route traffic needs different information than a human debugging why predictions aren’t working. "status": "ok" handles the first case; "model_loaded": false handles the second.
CORS Middleware
The frontend (running on a different origin than the API) needs CORS headers on API responses. FastAPI adds middleware declaratively:
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.allowed_origins, # ["http://localhost:5173"] in dev
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)Middleware wraps every request/response. CORS middleware reads the Origin header on the request and adds the Access-Control-Allow-Origin header to the response. The browser’s same-origin policy blocks cross-origin requests unless those headers are present.
allow_origins comes from settings rather than being hardcoded — in development it’s ["http://localhost:5173"] (SvelteKit’s dev server), in production it’s the deployed frontend URL. Hardcoding "*" is simpler but allows any origin to call the API, which is fine for a public read-only API and wrong for anything that involves authentication.
The middleware is added before the lifespan runs, which means CORS headers are on responses from the very first request — including the 503 responses that come back when the model isn’t loaded yet.
