Fast Api Pocket Book

Posted on by uplatzblog

FastAPI Pocket Book — Uplatz

50 deep-dive flashcards • Wide layout • Fewer scrolls • 20+ Interview Q&A • Readable Python snippets

Section 1 — Fundamentals

1) What is FastAPI?

FastAPI is a modern, high‑performance web framework for building APIs with Python 3.8+ based on Starlette (ASGI) and Pydantic for data validation. It generates OpenAPI/Swagger docs automatically and supports async/await natively.

pip install fastapi uvicorn[standard]
uvicorn app:app --reload

2) Why FastAPI? Strengths & Tradeoffs

Strengths: automatic validation & docs, excellent performance (uvicorn/uvloop), type hints for DX, async support. Tradeoffs: less batteries‑included than Django; you must pick ORM, auth, and tooling. Keep conventions to avoid sprawl.

# app.py
from fastapi import FastAPI
app = FastAPI()

3) ASGI vs WSGI

ASGI supports async concurrency, WebSockets, and background tasks. FastAPI apps run on ASGI servers (uvicorn, hypercorn). WSGI apps (Flask/Django classic) are sync‑only without adapters.

uvicorn app:app --workers 4 --http httptools --loop uvloop

4) Project Layout

Organize by feature: app/main.py, app/api/routers, app/core, app/models, app/services, tests.

app/
  main.py
  api/routers/users.py
  core/config.py
  db/session.py

5) Path Operations

Declare endpoints with Python type hints for params and responses.

from fastapi import FastAPI
app = FastAPI()
@app.get('/health')
async def health():
    return { 'ok': True }

6) Pydantic Models (v2‑style)

Use Pydantic models for request/response schemas; validation errors become JSON automatically.

from pydantic import BaseModel
class UserIn(BaseModel):
    email: str
    name: str | None = None

7) Response Models

Constrain output using response_model to avoid leaking fields.

class UserOut(BaseModel): id: int; email: str
@app.post('/users', response_model=UserOut)
async def create_user(u: UserIn): ...

8) Dependency Injection

Reuse logic (DB sessions, auth) via Depends. FastAPI resolves dependencies per‑request.

from fastapi import Depends
async def get_settings(): return { 'debug': False }
@app.get('/cfg')
async def cfg(s = Depends(get_settings)): return s

9) Settings & Env

Use Pydantic BaseSettings to read environment variables; fail fast if required keys are missing.

from pydantic_settings import BaseSettings
class Settings(BaseSettings): DB_URL: str
settings = Settings()

10) Q&A — “Is FastAPI production ready?”

Answer: Yes, with proper deployment: ASGI server (uvicorn/hypercorn), process manager (gunicorn), reverse proxy (nginx), observability, and a consistent architecture for auth/database.

Section 2 — Parameters, Bodies, Responses, Errors & Middleware

11) Path & Query Params

Type hints define validation and docs. Use Path, Query for extra constraints.

from fastapi import Path, Query
@app.get('/items/{id}')
async def get_item(id: int = Path(gt=0), q: str | None = Query(None, max_length=50)): ...

12) Request Body & Forms

Use models for JSON; use Form and UploadFile for forms/uploads.

from fastapi import Form, UploadFile, File
@app.post('/login')
async def login(username: str = Form(), password: str = Form()): ...
@app.post('/avatar')
async def avatar(file: UploadFile = File(...)): ...

13) Response Types

Return dicts, Pydantic models, or custom responses.

from fastapi.responses import JSONResponse, PlainTextResponse
@app.get('/txt', response_class=PlainTextResponse)
async def txt(): return 'hello'

14) Exception Handling

Raise HTTPException or register handlers for custom errors.

from fastapi import HTTPException
if not found: raise HTTPException(status_code=404, detail='Not found')

15) Global Error Handler

Customize error shapes and logging with exception handlers.

from fastapi import Request
from fastapi.responses import JSONResponse
@app.exception_handler(Exception)
async def all_errors(_: Request, exc: Exception):
    return JSONResponse({ 'error': str(exc) }, status_code=500)

16) Middleware & CORS

Add middleware for timing, auth, or CORS.

from starlette.middleware.cors import CORSMiddleware
app.add_middleware(CORSMiddleware, allow_origins=['https://app.example.com'], allow_headers=['*'], allow_methods=['*'], allow_credentials=True)

17) BackgroundTasks

For small, non‑critical async work after the response. For heavy jobs, use Celery/RQ.

from fastapi import BackgroundTasks
@app.post('/email')
async def send_email(bt: BackgroundTasks):
    bt.add_task(send_mail, 'hi')
    return {'queued': True}

18) Streaming & SSE

Stream large responses or use Server‑Sent Events.

from starlette.responses import StreamingResponse
async def gen():
    for i in range(5):
        yield f"data: {i}\n\n"
@app.get('/sse')
async def sse(): return StreamingResponse(gen(), media_type='text/event-stream')

19) WebSockets

FastAPI exposes Starlette WebSocket routes.

from fastapi import WebSocket
@app.websocket('/ws')
async def ws(ws: WebSocket):
    await ws.accept(); await ws.send_text('hello'); await ws.close()

20) Q&A — “BackgroundTasks vs Celery?”

Answer: Use BackgroundTasks for tiny, best‑effort jobs tied to a request. Use Celery/RQ/Arq for durable, retryable, scheduled, or heavy workloads.

Section 3 — Data, Auth, Caching & Dependencies

21) SQLAlchemy / SQLModel

Use SQLAlchemy 2.x or SQLModel for ORM. Manage sessions per‑request with dependencies and ensure proper cleanup.

from sqlalchemy.orm import Session
from .db import get_session
@app.get('/users')
async def list_users(db: Session = Depends(get_session)): ...

22) Database Session Pattern

Create a dependency that yields a session and closes it at the end.

from contextlib import contextmanager
@contextmanager
def session_scope():
    s = SessionLocal()
    try: yield s; s.commit()
    except: s.rollback(); raise
    finally: s.close()

23) NoSQL

For MongoDB, use Motor (async) with a dependency that returns a client/database; validate schemas with Pydantic models.

from motor.motor_asyncio import AsyncIOMotorClient

24) Pagination

Return items + next/prev cursors, not offset/limit for large data. Document in OpenAPI.

return { 'items': rows, 'next': next_cursor }

25) OAuth2 & JWT

Implement password flow or auth code with OpenID Connect; issue short‑lived access tokens and rotate refresh tokens.

from fastapi.security import OAuth2PasswordBearer
oauth2 = OAuth2PasswordBearer(tokenUrl='token')

26) Password Hashing

Use Passlib (bcrypt/argon2) and never store raw passwords.

from passlib.context import CryptContext
pwd = CryptContext(schemes=['bcrypt'])

27) Authorization (RBAC)

Create dependencies that enforce roles/permissions; return 403 on failure.

def require_role(role: str):
    def dep(user=Depends(current_user)):
        if role not in user.roles: raise HTTPException(403)
        return user
    return dep

28) Rate Limiting

Use Starlette middleware or external gateways; store counters in Redis and return 429 with Retry‑After.

# e.g. slowapi or custom middleware with redis

29) Caching

Cache hot reads in Redis with TTL; invalidate on writes. Include ETag/Cache‑Control for HTTP caching.

from fastapi import Response
resp = Response(content, headers={'Cache-Control': 'public, max-age=60'})

30) Q&A — “Where to put business logic?”

Answer: Keep it in services (or domain) modules called by routers. Routers translate HTTP↔domain; services handle rules, DB, external APIs.

Section 4 — Docs, Testing, Performance, Deployment & Observability

31) OpenAPI & Docs

FastAPI generates Swagger UI and ReDoc automatically; customize metadata and tags.

app = FastAPI(title='Shop API', version='1.0.0', openapi_tags=[{'name':'users'}])

32) Testing with httpx/pytest

Use httpx.AsyncClient and FastAPI’s lifespan for startup/shutdown.

import pytest
from httpx import AsyncClient
@pytest.mark.asyncio
async def test_health():
    async with AsyncClient(app=app, base_url='http://test') as ac:
        r = await ac.get('/health')
        assert r.status_code == 200

33) Dependency Overrides in Tests

Override Depends providers for fakes/mocks.

app.dependency_overrides[get_session] = fake_session

34) Performance Tips

Avoid blocking calls in async routes; use threadpools for CPU‑bound tasks; enable uvloop/httptools; compress & cache responses.

import anyio
result = await anyio.to_thread.run_sync(cpu_heavy)

35) Gunicorn + UvicornWorkers

Run multiple worker processes under Gunicorn for multi‑core scaling.

gunicorn app:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000

36) Lifespan Events

Initialize DB pools/clients on startup and close at shutdown using the lifespan context.

from contextlib import asynccontextmanager
@asynccontextmanager
async def lifespan(app):
    yield
app = FastAPI(lifespan=lifespan)

37) Security Headers

Add CSP, HSTS, and other headers via middleware when serving pages; APIs typically set CORS, cache, and auth headers.

from starlette.middleware import Middleware

38) Logging

Structure logs (JSON), include request IDs, and capture exceptions. Configure uvicorn/gunicorn log levels consistently.

import logging, sys
logging.basicConfig(stream=sys.stdout, level=logging.INFO)

39) Observability

Export Prometheus metrics and OpenTelemetry traces; tag by route/status for SLOs.

# use prometheus-fastapi-instrumentator or custom middleware

40) Q&A — “How many workers/threads?”

Answer: Start with workers ≈ CPU cores, keep threads low (async handles concurrency). Load test and adjust for latency/SLOs; prefer horizontal scaling.

Section 5 — Deployment, Security, Patterns, Pitfalls & Interview Q&A

41) Dockerfile

Build slim images and pin Python versions.

FROM python:3.12-slim
WORKDIR /app
COPY pyproject.toml requirements.txt ./
RUN pip install -r requirements.txt
COPY . .
CMD ["gunicorn","app:app","-k","uvicorn.workers.UvicornWorker","-w","4","-b","0.0.0.0:8000"]

42) Health & Readiness

Provide /health (shallow) and /ready (checks DB/queues) for orchestrators.

@app.get('/ready')
async def ready(): return {'db': 'ok', 'queue': 'ok'}

43) Security Basics

Validate inputs, limit body size, sanitize outputs, require HTTPS, rotate secrets, and enforce least privilege.

from fastapi import Request
@app.middleware('http')
async def size_limit(req: Request, call_next):
    if int(req.headers.get('content-length', 0)) > 1_000_000:
        return JSONResponse({'error': 'too large'}, 413)
    return await call_next(req)

44) File Upload Safety

Validate filename/content‑type, store outside web root, and scan untrusted files.

dest = secure_path(filename)

45) Task Queues

Offload slow/retryable work to Celery/RQ/Arq; use idempotency keys and DLQs.

# celery -A worker.celery worker -l info

46) Production Checklist

  • Type‑hinted routers & response models
  • Dependency‑based DB/session/auth
  • HTTP timeouts, rate limits, CORS
  • Health/ready endpoints; graceful shutdown
  • Structured logs, metrics, tracing
  • Load tests; autoscaling & runbooks

47) Common Pitfalls

Blocking calls in async routes, forgetting dependency overrides in tests, leaking DB sessions, not setting CORS/HTTPS, and exposing sensitive fields by skipping response_model.

48) Example App Skeleton

Split creation and running for testability.

# app/main.py
from fastapi import FastAPI
from .api.routers import users

def create_app() -> FastAPI:
    app = FastAPI()
    app.include_router(users.router, prefix='/users', tags=['users'])
    return app

app = create_app()

49) Versioning & Deprecation

Version via path (/v1) or headers; publish timelines and deprecation headers.

from fastapi import Response
Response(headers={'Deprecation':'true','Sunset':'Wed, 31 Dec 2025 23:59:59 GMT'})

50) Interview Q&A — 20 Practical Questions (FastAPI‑focused)

1) Why FastAPI for microservices? Strong typing, speed, and auto‑docs accelerate teams.

2) ASGI advantages? Concurrency, WebSockets, streaming; works with uvicorn/hypercorn.

3) Where to put DB session management? In dependencies that yield/close per request.

4) How to handle validation? Pydantic models + type hints; errors are automatic.

5) Response model purpose? Control output fields and docs; prevent data leaks.

6) Auth options? OAuth2/OIDC with JWT; cookie sessions for web apps via middleware.

7) Background tasks vs queues? BackgroundTasks for quick follow‑ups; Celery for durable jobs.

8) Testing tools? pytest + httpx AsyncClient; dependency overrides; fixtures for DB.

9) Performance levers? uvloop/httptools, caching, pagination, avoid blocking code.

10) CORS configuration? CORSMiddleware with explicit origins/headers.

11) OpenAPI customization? Title, version, tags, and examples on models.

12) File uploads safely? Validate type/size, store outside root, scan.

13) Pagination pattern? Cursor‑based for scale; include links/tokens.

14) Observability stack? Structured logs, Prometheus metrics, OpenTelemetry traces.

15) Deploy on k8s? Gunicorn UvicornWorkers, readiness/liveness probes, HPA by CPU/latency.

16) Handle partial failures? Timeouts, retries with backoff, circuit breakers.

17) Secure config? BaseSettings + secret manager; do not commit secrets.

18) Prevent over‑fetch? response_model + select only needed columns; paginate.

19) WebSockets auth? Query tokens/cookies at websocket.accept(); validate before joining rooms.

20) When not to use FastAPI? Heavy server‑side templating or complex admin UIs — consider Django.