Advanced FastAPI Patterns
Lifespan (startup/shutdown)
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup
await db_pool.start()
await redis.connect()
logger.info("Application started")
yield # app is running
# Shutdown
await db_pool.close()
await redis.disconnect()
logger.info("Application stopped")
app = FastAPI(lifespan=lifespan)
OpenAPI Customization
from fastapi.openapi.utils import get_openapi
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
schema = get_openapi(
title="My API",
version="1.0.0",
description="Production API documentation",
routes=app.routes,
)
# Add security scheme
schema["components"]["securitySchemes"] = {
"BearerAuth": {"type": "http", "scheme": "bearer", "bearerFormat": "JWT"}
}
schema["security"] = [{"BearerAuth": []}]
app.openapi_schema = schema
return schema
app.openapi = custom_openapi
File Uploads
from fastapi import UploadFile, File
import aiofiles
ALLOWED_TYPES = {"image/jpeg", "image/png", "application/pdf"}
MAX_FILE_SIZE = 10 * 1024 * 1024 # 10MB
@router.post("/upload")
async def upload_file(
file: UploadFile = File(...),
current_user: User = Depends(get_current_user),
) -> dict:
# Validate content type
if file.content_type not in ALLOWED_TYPES:
raise HTTPException(400, f"File type {file.content_type} not allowed")
# Stream to check size without loading all into memory
contents = b""
async for chunk in file:
contents += chunk
if len(contents) > MAX_FILE_SIZE:
raise HTTPException(413, "File too large")
# Save
dest = Path(f"uploads/{uuid4()}_{file.filename}")
async with aiofiles.open(dest, "wb") as out:
await out.write(contents)
return {"filename": file.filename, "size": len(contents), "path": str(dest)}
Middleware Stack
from fastapi.middleware.cors import CORSMiddleware
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from fastapi.middleware.gzip import GZipMiddleware
# Order matters: outermost first
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(TrustedHostMiddleware, allowed_hosts=["api.example.com", "localhost"])
Custom Exception Handling
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
@app.exception_handler(RequestValidationError)
async def validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
return JSONResponse(
status_code=422,
content={
"error": "validation_error",
"details": [
{"field": ".".join(str(loc) for loc in err["loc"]), "message": err["msg"]}
for err in exc.errors()
],
},
)
@app.exception_handler(StarletteHTTPException)
async def http_error_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse:
return JSONResponse(
status_code=exc.status_code,
content={"error": exc.detail},
)
Mattakushi432