move
This commit is contained in:
John Lancaster
@@ -0,0 +1,238 @@
|
||||
# FastAPI Best Practices
|
||||
|
||||
Source: https://fastapi.tiangolo.com/deployment/ | https://fastapi.tiangolo.com/advanced/events/
|
||||
|
||||
---
|
||||
|
||||
## App Factory Pattern
|
||||
|
||||
Always use `create_app()` — it makes the app testable (inject a test `Settings`) and avoids module-level side effects.
|
||||
|
||||
```python
|
||||
# src/my_app/main.py
|
||||
from contextlib import asynccontextmanager
|
||||
from fastapi import FastAPI
|
||||
from my_app.config import Settings
|
||||
from my_app.api import health, v1
|
||||
|
||||
def create_app(settings: Settings | None = None) -> FastAPI:
|
||||
if settings is None:
|
||||
settings = Settings()
|
||||
|
||||
@asynccontextmanager
|
||||
async def lifespan(app: FastAPI):
|
||||
# --- startup ---
|
||||
app.state.settings = settings
|
||||
# Open DB pool, warm caches, etc.
|
||||
yield
|
||||
# --- shutdown ---
|
||||
# Close DB pool, flush buffers, etc.
|
||||
|
||||
app = FastAPI(
|
||||
title=settings.app_name,
|
||||
version="1.0.0",
|
||||
lifespan=lifespan,
|
||||
docs_url="/docs" if settings.debug else None,
|
||||
redoc_url=None,
|
||||
)
|
||||
|
||||
app.include_router(health.router)
|
||||
app.include_router(v1.router, prefix="/api/v1")
|
||||
return app
|
||||
|
||||
# Module-level instance for uvicorn
|
||||
app = create_app()
|
||||
```
|
||||
|
||||
**Never use `@app.on_event("startup")` / `@app.on_event("shutdown")`** — these are deprecated. The `asynccontextmanager` lifespan is the canonical approach since FastAPI 0.95.
|
||||
|
||||
---
|
||||
|
||||
## Router Organization
|
||||
|
||||
```
|
||||
src/my_app/api/
|
||||
├── __init__.py
|
||||
├── health.py # GET /healthz — no auth, no versioning
|
||||
├── deps.py # Shared Depends() factories
|
||||
└── v1/
|
||||
├── __init__.py # APIRouter with prefix="/v1"
|
||||
├── items.py
|
||||
└── users.py
|
||||
```
|
||||
|
||||
Each router file:
|
||||
|
||||
```python
|
||||
from fastapi import APIRouter
|
||||
|
||||
router = APIRouter(prefix="/items", tags=["items"])
|
||||
|
||||
@router.get("/")
|
||||
async def list_items() -> list[Item]:
|
||||
...
|
||||
```
|
||||
|
||||
Root registration:
|
||||
|
||||
```python
|
||||
from fastapi import APIRouter
|
||||
from my_app.api.v1 import items, users
|
||||
|
||||
router = APIRouter()
|
||||
router.include_router(items.router)
|
||||
router.include_router(users.router)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Pydantic Settings (Configuration)
|
||||
|
||||
```python
|
||||
# src/my_app/config.py
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from functools import lru_cache
|
||||
|
||||
class Settings(BaseSettings):
|
||||
model_config = SettingsConfigDict(
|
||||
env_file=".env",
|
||||
env_file_encoding="utf-8",
|
||||
extra="ignore",
|
||||
)
|
||||
|
||||
app_name: str = "My App"
|
||||
debug: bool = False
|
||||
host: str = "0.0.0.0"
|
||||
port: int = 8000
|
||||
log_level: str = "info"
|
||||
# Add DB URL, secret keys, etc. here — never hardcode
|
||||
# database_url: str # required — will raise if missing
|
||||
|
||||
@lru_cache
|
||||
def get_settings() -> Settings:
|
||||
return Settings()
|
||||
```
|
||||
|
||||
Use `lru_cache` so the `.env` file is read once. In tests, override with:
|
||||
|
||||
```python
|
||||
from my_app.config import get_settings
|
||||
from my_app.main import create_app
|
||||
|
||||
app = create_app(settings=Settings(debug=True, database_url="sqlite://"))
|
||||
```
|
||||
|
||||
**Never** import `settings` as a module-level singleton — it prevents test overrides.
|
||||
|
||||
---
|
||||
|
||||
## Health Endpoint
|
||||
|
||||
```python
|
||||
# src/my_app/api/health.py
|
||||
from fastapi import APIRouter
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
router = APIRouter(tags=["ops"])
|
||||
|
||||
@router.get("/healthz", include_in_schema=False)
|
||||
async def health() -> JSONResponse:
|
||||
"""Kubernetes/Docker liveness probe. No auth. No versioning."""
|
||||
return JSONResponse({"status": "ok"})
|
||||
|
||||
@router.get("/readyz", include_in_schema=False)
|
||||
async def readiness(request: Request) -> JSONResponse:
|
||||
"""Readiness probe — check DB connectivity, etc."""
|
||||
# Example: await request.app.state.db.execute("SELECT 1")
|
||||
return JSONResponse({"status": "ready"})
|
||||
```
|
||||
|
||||
Rules:
|
||||
- No authentication required on `/healthz` and `/readyz`.
|
||||
- `/healthz` — liveness: can the process respond?
|
||||
- `/readyz` — readiness: are dependencies available?
|
||||
- Keep them on the root path (not `/api/v1/healthz`).
|
||||
|
||||
---
|
||||
|
||||
## Dependency Injection
|
||||
|
||||
Use `Depends()` to share resources from app state:
|
||||
|
||||
```python
|
||||
# src/my_app/api/deps.py
|
||||
from fastapi import Depends, Request
|
||||
from my_app.config import Settings
|
||||
|
||||
def get_settings(request: Request) -> Settings:
|
||||
return request.app.state.settings
|
||||
|
||||
SettingsDep = Annotated[Settings, Depends(get_settings)]
|
||||
```
|
||||
|
||||
In route handlers:
|
||||
|
||||
```python
|
||||
@router.get("/config")
|
||||
async def show_config(settings: SettingsDep) -> dict:
|
||||
return {"debug": settings.debug}
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Error Handling
|
||||
|
||||
Register a global exception handler for unhandled errors:
|
||||
|
||||
```python
|
||||
from fastapi import Request
|
||||
from fastapi.responses import JSONResponse
|
||||
|
||||
@app.exception_handler(Exception)
|
||||
async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
|
||||
# Log the exception — never expose internal details to clients
|
||||
logger.exception("Unhandled error", exc_info=exc)
|
||||
return JSONResponse(status_code=500, content={"detail": "Internal server error"})
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## CORS
|
||||
|
||||
Add CORS middleware only when needed (e.g., browser clients from a different origin):
|
||||
|
||||
```python
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=settings.cors_origins, # list from config, never ["*"] in prod
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
)
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## Response Models
|
||||
|
||||
Always declare `response_model` or return type annotations — FastAPI uses them for OpenAPI docs and response validation:
|
||||
|
||||
```python
|
||||
@router.post("/items/", response_model=ItemOut, status_code=201)
|
||||
async def create_item(item: ItemIn) -> ItemOut:
|
||||
...
|
||||
```
|
||||
|
||||
Use separate `In` / `Out` models when the write shape differs from the read shape (e.g., password hashing, computed fields).
|
||||
|
||||
---
|
||||
|
||||
## Security Checklist
|
||||
|
||||
- Never expose `docs_url` in production (set `docs_url=None` when `not settings.debug`).
|
||||
- Validate all user input with Pydantic models — never pass raw request data to DB queries.
|
||||
- Use `SecretStr` for passwords and API keys in `Settings`.
|
||||
- Apply authentication globally via middleware or `app.include_router(..., dependencies=[Depends(verify_token)])`.
|
||||
- Add `X-Content-Type-Options: nosniff` and `X-Frame-Options: DENY` headers in production middleware.
|
||||
Reference in New Issue
Block a user