FastAPI: High-Performance Python Web Framework with Type Safety
Async/Await and Performance
FastAPI uses async/await (Python 3.6+). Traditional Flask: requests processed sequentially (request 1 takes 1 second, request 2 waits 1 second, total 2 seconds). Async FastAPI: both requests processed concurrently (await I/O, serve next request, resume when I/O done, total ~1 second). ASGI server (Uvicorn): event loop manages async tasks. Example: @app.get("/items/{id}"), async def get_item(id: int) { data = await db.query("SELECT * FROM items WHERE id = ?", id); return data }. Uvicorn uses async event loop (single-threaded, non-blocking I/O). Worker processes: Uvicorn workers scale horizontally (4 workers on 4-core machine = 4 concurrent requests per core, 16 total). Performance: FastAPI ~500-1000 requests/second (simple endpoint), Flask ~100-300 requests/second (10x difference). Database connections: await db.fetch (async driver), vs blocking SQLAlchemy (blocks event loop). Starlette: ASGI framework FastAPI built on (provides routing, middleware, background tasks). Pydantic: type validation, automatic OpenAPI schema generation from types.
Type Hints and Validation
Pydantic models: automatic validation and serialization. Example: class Item(BaseModel): name: str, price: float, in_stock: bool = True. Validation: Item(name="Widget", price=9.99) succeeds, Item(name="Widget", price="invalid") raises ValidationError (price not float). FastAPI extracts types from model: @app.post("/items/"), async def create_item(item: Item): (body auto-parsed, validated against Item schema). Required fields: name, price mandatory. Optional fields: in_stock defaults to True. Type validation: price: float (enforced), min_length, max_length, regex patterns. Example: name: str = Field(..., min_length=1, max_length=100) (name required, 1-100 characters). Field validators: @validator('name'), def name_must_not_be_empty(cls, v) { if not v: raise ValueError('empty'); return v }. Custom types: EmailStr (validates email), HttpUrl (validates URL). Automatic OpenAPI schema generation: types → Swagger documentation (GET /docs shows interactive API docs).
Dependency Injection and Middleware
FastAPI dependency system: share common logic across endpoints. Example: Depends(get_current_user) authenticates request. async def get_current_user(token: str = Header(...)): user = await verify_token(token); if not user: raise HTTPException(401); return user. Multiple endpoints: @app.get("/profile"), async def get_profile(user: User = Depends(get_current_user)): return user.profile. Dependency runs once per request (cached within request). Nested dependencies: get_current_user calls verify_token, FastAPI caches both. Middleware: add custom request/response handling. Example: @app.middleware("http"), async def add_process_time_header(request: Request, call_next): start = time.time(); response = await call_next(request); response.headers["X-Process-Time"] = str(time.time() - start); return response. Middleware chain: request → middleware 1 → middleware 2 → endpoint → middleware 2 → middleware 1 → response. Authentication: JWT middleware decodes token, stores user in request state. CORS middleware: allow cross-origin requests. Compression: gzip response (30-80% size reduction).
Background Tasks and WebSockets
Background tasks: execute after response sent. Example: @app.post("/email/"), async def send_email(email: str): background_tasks.add_task(send_email_background, email); return {"message": "Email queued"}. Task runs after response returns (client sees response immediately, email sends in background). Use case: send notification, log event, update cache. Async background tasks: within same process (simple), but blocks other requests if many tasks. Job queue (Celery, RQ): offload tasks to separate workers (production recommendation). WebSockets: bidirectional communication. @app.websocket("/ws/{client_id}"), async def websocket_endpoint(websocket: WebSocket, client_id: str): await websocket.accept(); while True: data = await websocket.receive_text(); await websocket.send_text(f"You said: {data}"). Client connects, server sends message, client receives, sends response (back-and-forth). Use case: real-time chat, live notifications, collaborative editing. Broadcasting: server sends message to multiple connected clients (manage connection list).
Database Integration and ORM
SQLAlchemy async: async-native ORM. Example: from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, engine = create_async_engine("postgresql+asyncpg://..."). async with AsyncSession(engine) as session: result = await session.execute(select(User).where(User.id == 1)); user = result.scalar_one(). Async driver: asyncpg (PostgreSQL), motor (MongoDB). Benefits: non-blocking I/O, handles 100s of concurrent connections (blocking driver would need 100s of threads). Alembic: database migration tool. alembic revision --autogenerate -m "add user table" creates migration script. alembic upgrade head applies migrations. Transactions: async with session.begin(): (commits on success, rolls back on error). Relationship loading: eager vs lazy loading. selectinload: join in SQL, fetch related data (1 query). subqueryload: separate subquery (2 queries). selectinload faster but uses more memory. Connection pooling: pool_size=20 (max 20 connections), max_overflow=10 (additional connections if pool exhausted).
Testing and Deployment
Testing: TestClient wraps FastAPI app. Example: from fastapi.testclient import TestClient, client = TestClient(app), response = client.get("/items/1"), assert response.status_code == 200. Test structure: @pytest.fixture, async def test_get_item(): response = client.get("/items/1"); assert response.json() == {"name": "Item 1", "price": 9.99}. Mocking: monkeypatch replaces dependencies. @pytest.fixture, def mock_db(monkeypatch): monkeypatch.setattr(app.dependency_overrides[get_db], mock_db_func). Code coverage: pytest --cov=app (target 80%+ coverage). Deployment: Gunicorn + Uvicorn. gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app (4 workers). Docker: FROM python:3.11, COPY requirements.txt ., RUN pip install -r requirements.txt, COPY . ., CMD ["uvicorn", "main:app", "--host", "0.0.0.0"]. Kubernetes: deployment.yaml defines pod replicas (auto-scale based on CPU/memory). Cloud platforms: AWS EC2, Google Cloud Run, Heroku support FastAPI (zero-config deployment on some).
Performance Optimization and Monitoring
Profiling: profile endpoint execution time. from pyinstrument import Profiler, prof = Profiler(), prof.start(), function_call(), prof.stop(), print(prof.output_text()). Identifies bottlenecks (database queries, CPU-heavy operations). Caching: from functools import lru_cache, @lru_cache (memoize function results). Redis cache: from aioredis import Redis, value = await redis.get(key) (check cache before database). Cache invalidation: delete key after database update. Rate limiting: from slowapi import Limiter, @limiter.limit("5/minute") (5 requests per minute per IP). Metrics: Prometheus exports metrics (/metrics). from prometheus_client import Counter, counter = Counter('requests', 'Total requests'), counter.inc() (increment on each request). Example metrics: request count, latency histogram (50th/95th/99th percentiles), error rate. Alerting: Prometheus alert rules trigger on metrics (error_rate > 5% → alert). Health checks: GET /health returns 200 if service healthy (Kubernetes uses for liveness probes). Structured logging: import structlog, log.msg("request", method="GET", path="/items", status=200, duration_ms=45) (JSON format, easy parsing).