home / skills / yonatangross / orchestkit / fastapi-advanced
/plugins/ork/skills/fastapi-advanced
This skill helps you implement advanced FastAPI patterns including lifespan, dependency injection, middleware, and settings for robust backends.
npx playbooks add skill yonatangross/orchestkit --skill fastapi-advancedReview the files below or copy the command above to add this skill to your agents.
---
name: fastapi-advanced
description: FastAPI 2026 advanced patterns including lifespan, dependencies, middleware, and Pydantic settings. Use when configuring FastAPI lifespan events, creating dependency injection, building Starlette middleware, or managing async Python services with uvicorn.
context: fork
agent: backend-system-architect
version: 1.0.0
tags: [fastapi, python, async, middleware, dependencies, 2026]
author: OrchestKit
user-invocable: false
---
# FastAPI Advanced Patterns (2026)
Production-ready FastAPI patterns for modern Python backends.
## Lifespan Management (2026)
### Modern Lifespan Context Manager
```python
from contextlib import asynccontextmanager
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
import redis.asyncio as redis
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifespan with resource management."""
# Startup
app.state.db_engine = create_async_engine(
settings.database_url,
pool_size=5,
max_overflow=10,
)
app.state.redis = redis.from_url(settings.redis_url)
# Health check connections
async with app.state.db_engine.connect() as conn:
await conn.execute(text("SELECT 1"))
await app.state.redis.ping()
yield # Application runs
# Shutdown
await app.state.db_engine.dispose()
await app.state.redis.close()
app = FastAPI(lifespan=lifespan)
```
### Lifespan with Services
```python
from app.services import EmbeddingsService, LLMService
@asynccontextmanager
async def lifespan(app: FastAPI):
# Initialize services
app.state.embeddings = EmbeddingsService(
model=settings.embedding_model,
batch_size=100,
)
app.state.llm = LLMService(
providers=["openai", "anthropic"],
default="anthropic",
)
# Warm up models (optional)
await app.state.embeddings.warmup()
yield
# Cleanup
await app.state.embeddings.close()
await app.state.llm.close()
```
## Dependency Injection Patterns
### Database Session
```python
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi import Depends, Request
async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]:
"""Yield database session from app state."""
async with AsyncSession(
request.app.state.db_engine,
expire_on_commit=False,
) as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
```
### Service Dependencies
```python
from functools import lru_cache
class AnalysisService:
def __init__(
self,
db: AsyncSession,
embeddings: EmbeddingsService,
llm: LLMService,
):
self.db = db
self.embeddings = embeddings
self.llm = llm
def get_analysis_service(
db: AsyncSession = Depends(get_db),
request: Request = None,
) -> AnalysisService:
return AnalysisService(
db=db,
embeddings=request.app.state.embeddings,
llm=request.app.state.llm,
)
@router.post("/analyses")
async def create_analysis(
data: AnalysisCreate,
service: AnalysisService = Depends(get_analysis_service),
):
return await service.create(data)
```
### Cached Dependencies
```python
from functools import lru_cache
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
redis_url: str
api_key: str
model_config = {"env_file": ".env"}
@lru_cache
def get_settings() -> Settings:
return Settings()
# Usage in dependencies
def get_db_url(settings: Settings = Depends(get_settings)) -> str:
return settings.database_url
```
### Authenticated User
```python
from fastapi import Depends, HTTPException, Security
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Security(security),
db: AsyncSession = Depends(get_db),
) -> User:
token = credentials.credentials
payload = decode_jwt(token)
user = await db.get(User, payload["sub"])
if not user:
raise HTTPException(401, "Invalid credentials")
return user
async def get_admin_user(
user: User = Depends(get_current_user),
) -> User:
if not user.is_admin:
raise HTTPException(403, "Admin access required")
return user
```
## Middleware Patterns
### Request ID Middleware
```python
import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
class RequestIDMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
request.state.request_id = request_id
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
return response
app.add_middleware(RequestIDMiddleware)
```
### Timing Middleware
```python
import time
from starlette.middleware.base import BaseHTTPMiddleware
class TimingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
start = time.perf_counter()
response = await call_next(request)
duration = time.perf_counter() - start
response.headers["X-Response-Time"] = f"{duration:.3f}s"
return response
```
### Structured Logging Middleware
```python
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
logger = structlog.get_logger()
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
log = logger.bind(
request_id=getattr(request.state, "request_id", None),
method=request.method,
path=request.url.path,
)
try:
response = await call_next(request)
log.info(
"request_completed",
status_code=response.status_code,
)
return response
except Exception as exc:
log.exception("request_failed", error=str(exc))
raise
```
### CORS Configuration
```python
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
allow_headers=["*"],
expose_headers=["X-Request-ID", "X-Response-Time"],
)
```
## Settings with Pydantic
```python
from pydantic import Field, field_validator, PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
case_sensitive=False,
)
# Database
database_url: PostgresDsn
db_pool_size: int = Field(default=5, ge=1, le=20)
db_max_overflow: int = Field(default=10, ge=0, le=50)
# Redis
redis_url: str = "redis://localhost:6379"
# API
api_key: str = Field(min_length=32)
debug: bool = False
# LLM
openai_api_key: str | None = None
anthropic_api_key: str | None = None
@field_validator("database_url", mode="before")
@classmethod
def validate_database_url(cls, v: str) -> str:
if v and "+asyncpg" not in v:
return v.replace("postgresql://", "postgresql+asyncpg://")
return v
@property
def async_database_url(self) -> str:
return str(self.database_url)
```
## Exception Handlers
```python
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError
from app.core.exceptions import ProblemException
@app.exception_handler(ProblemException)
async def problem_exception_handler(request: Request, exc: ProblemException):
return JSONResponse(
status_code=exc.status_code,
content=exc.to_problem_detail(),
media_type="application/problem+json",
)
@app.exception_handler(IntegrityError)
async def integrity_error_handler(request: Request, exc: IntegrityError):
return JSONResponse(
status_code=409,
content={
"type": "https://api.example.com/problems/conflict",
"title": "Conflict",
"status": 409,
"detail": "Resource already exists or constraint violated",
},
media_type="application/problem+json",
)
```
## Response Optimization
```python
from fastapi.responses import ORJSONResponse
# Use orjson for faster JSON serialization
app = FastAPI(default_response_class=ORJSONResponse)
# Streaming response
from fastapi.responses import StreamingResponse
@router.get("/export")
async def export_data():
async def generate():
async for chunk in fetch_large_dataset():
yield json.dumps(chunk) + "\n"
return StreamingResponse(
generate(),
media_type="application/x-ndjson",
)
```
## Health Checks
```python
from fastapi import APIRouter
health_router = APIRouter(tags=["health"])
@health_router.get("/health")
async def health_check(request: Request):
checks = {}
# Database
try:
async with request.app.state.db_engine.connect() as conn:
await conn.execute(text("SELECT 1"))
checks["database"] = "healthy"
except Exception as e:
checks["database"] = f"unhealthy: {e}"
# Redis
try:
await request.app.state.redis.ping()
checks["redis"] = "healthy"
except Exception as e:
checks["redis"] = f"unhealthy: {e}"
status = "healthy" if all(v == "healthy" for v in checks.values()) else "unhealthy"
return {"status": status, "checks": checks}
```
## Anti-Patterns (FORBIDDEN)
```python
# NEVER use global state
db_session = None # Global mutable state!
# NEVER block the event loop
def sync_db_query(): # Blocking in async context!
return session.query(User).all()
# NEVER skip dependency injection
@router.get("/users")
async def get_users():
db = create_session() # Creating session in route!
return db.query(User).all()
# NEVER ignore lifespan cleanup
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.pool = create_pool()
yield
# Missing cleanup! Pool never closed
```
## Key Decisions
| Decision | Recommendation |
|----------|----------------|
| Lifespan | Use `asynccontextmanager` (not events) |
| Dependencies | Class-based services with DI |
| Settings | Pydantic Settings with `.env` |
| Response | ORJSONResponse for performance |
| Middleware | Order: CORS → RequestID → Timing → Logging |
| Health | Check all critical dependencies |
## Available Scripts
- **`scripts/create-fastapi-app.md`** - Context-aware FastAPI application generator
- Auto-detects: Python version, database type, Redis usage, project structure
- Usage: `/create-fastapi-app [app-name]`
- Uses `$ARGUMENTS` and `!command` for project-specific configuration
- Generates production-ready app with detected dependencies
- **`assets/fastapi-app-template.py`** - Static FastAPI application template
## Related Skills
- `clean-architecture` - Service layer patterns
- `database-schema-designer` - SQLAlchemy models
- `observability-monitoring` - Logging and metrics
## Capability Details
### lifespan
**Keywords:** lifespan, startup, shutdown, asynccontextmanager
**Solves:**
- FastAPI startup/shutdown
- Resource management in FastAPI
### dependencies
**Keywords:** dependency injection, Depends, get_db, service dependency
**Solves:**
- FastAPI dependency injection patterns
- Reusable dependencies
### middleware
**Keywords:** middleware, request id, timing, cors, logging middleware
**Solves:**
- Custom FastAPI middleware
- Request/response interceptors
### settings
**Keywords:** settings, pydantic settings, env, configuration
**Solves:**
- FastAPI configuration management
- Environment variables
### health-checks
**Keywords:** health check, readiness, liveness, health endpoint
**Solves:**
- Kubernetes health checks
- Service health monitoring
This skill packages production-ready FastAPI 2026 patterns for building reliable async Python backends. It focuses on lifespan management, dependency injection, middleware, Pydantic settings, health checks, and response optimizations to run safely with uvicorn and modern LLM services.
I provide reusable patterns: an asynccontextmanager-based lifespan that initializes and cleans up DB engines, Redis, and LLM/embeddings services. Dependency factories yield scoped DB sessions and class-based services wired to app.state. Middleware examples include request IDs, timing, and structured logging. Settings use Pydantic Settings with validators and environment file support.
How do I ensure DB URLs work with async drivers?
Use a Pydantic field validator to normalize database_url (append +asyncpg) and expose an async_database_url property.
Where should I put long-lived clients like Redis or LLMs?
Initialize them in the lifespan manager and attach to app.state; access them in dependencies via request.app.state.