FastAPI, Pydantic & async correctness

What is ASGI, and how does it differ from WSGI?

WSGI is a synchronous, one-request-per-callable interface: app(environ, start_response). It cannot express long-lived connections. ASGI is async and event-based — await app(scope, receive, send) — with scope["type"] distinguishing http, websocket and lifespan, and receive/send streaming message dicts.

That is why FastAPI can do WebSockets, server-sent events, streaming uploads and background lifespan setup, and why a single process can hold thousands of open connections. FastAPI is Starlette (ASGI toolkit) + Pydantic (validation) + automatic OpenAPI.

def vs async def in a FastAPI path operation — what actually happens?

• async def → runs on the event loop. If you block in it, you block every other request in that worker.
• def (sync) → FastAPI runs it in a threadpool (run_in_threadpool, 40 threads by default), so blocking is safe but concurrency is bounded by the pool.

@app.get("/bad")
async def bad():
    return requests.get(url).json()        # blocks the loop — throughput collapses

@app.get("/ok-sync")
def ok_sync():
    return requests.get(url).json()        # fine: runs in a thread

@app.get("/best")
async def best():
    r = await client.get(url)              # httpx.AsyncClient, created once at startup
    return r.json()

The rule to state: if any part of the handler blocks, use def; if everything awaits, use async def. A half-async handler is the worst of both.

Explain FastAPI's dependency injection. Why is it more than a decorator?

Depends builds a DAG of callables resolved per request, with results cached within that request (so a get_current_user used by three sub-dependencies runs once), support for sync and async, generator dependencies with teardown, and automatic inclusion in the OpenAPI schema.

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        yield session                       # code after yield runs after the response
                                            # (and on exception, with the exception raised)

async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: Annotated[AsyncSession, Depends(get_db)],
) -> User:
    try:
        payload = jwt.decode(token, SECRET, algorithms=["HS256"])
    except JWTError:
        raise HTTPException(401, "invalid token", headers={"WWW-Authenticate": "Bearer"})
    user = await db.get(User, payload["sub"])
    if user is None or not user.active:
        raise HTTPException(401, "invalid token")
    return user

@app.get("/me")
async def me(user: Annotated[User, Depends(get_current_user)]) -> UserOut:
    return user

Router- and app-level dependencies (APIRouter(dependencies=[Depends(require_admin)])) enforce cross-cutting rules without touching each handler — the right way to do authorisation on a whole section.

How does Pydantic v2 validation work, and what changed from v1?

v2's core is pydantic-core, written in Rust — typically 5-20x faster. API changes worth naming: BaseSettings moved to pydantic-settings; @validator/@root_validator → @field_validator/@model_validator; .dict()/.json() → .model_dump()/.model_dump_json(); Config class → model_config = ConfigDict(...); parse_obj → model_validate.

class UserCreate(BaseModel):
    model_config = ConfigDict(str_strip_whitespace=True, extra="forbid")
    email: EmailStr
    password: SecretStr = Field(min_length=12)
    age: int | None = Field(default=None, ge=13, le=130)

    @field_validator("email")
    @classmethod
    def lowercase(cls, v: str) -> str: return v.lower()

    @model_validator(mode="after")
    def check(self) -> "UserCreate":
        if self.password.get_secret_value() in self.email: raise ValueError("too guessable")
        return self

extra="forbid" is a security-relevant default: it rejects unexpected fields instead of silently ignoring them.

Why use separate input and output models?

Because the fields a client may send are not the fields you should return. One model for both leaks password_hash, is_admin and internal ids, and lets clients set fields they should not (mass assignment).

class UserIn(BaseModel):  email: EmailStr; password: SecretStr
class UserOut(BaseModel): id: int; email: EmailStr; created_at: datetime
                          model_config = ConfigDict(from_attributes=True)

@app.post("/users", response_model=UserOut, status_code=201)
async def create(payload: UserIn, db=Depends(get_db)) -> Any: ...

response_model also filters the response, so an accidental extra attribute never reaches the wire. That belt-and-braces behaviour is the point.

BackgroundTasks vs Celery — when is each right?

BackgroundTasks runs after the response in the same process: no durability, no retries, lost on restart or crash, and it competes for the same event loop or threadpool. Fine for a fire-and-forget log write or a cache warm.

Use Celery/ARQ/Dramatiq/a queue when the work must survive a restart, needs retries or scheduling, is CPU-heavy, or must scale independently. The test: "if this task silently disappeared, would anyone care?" If yes, it needs a real queue.

How do you manage startup and shutdown resources?

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10.0)     # ONE pooled client for the process
    app.state.redis = await aioredis.from_url(REDIS_URL)
    yield
    await app.state.http.aclose()
    await app.state.redis.close()

app = FastAPI(lifespan=lifespan)

Creating an httpx.AsyncClient (or a DB engine) per request is a classic performance bug — you lose connection pooling and pay a TLS handshake every time, and you eventually exhaust file descriptors. The on_event("startup") decorators are deprecated in favour of lifespan.

How do you handle errors consistently?

class DomainError(Exception):
    def __init__(self, code: str, status: int = 400): self.code, self.status = code, status

@app.exception_handler(DomainError)
async def domain_handler(request: Request, exc: DomainError):
    return JSONResponse(status_code=exc.status,
                        content={"code": exc.code, "request_id": request.state.request_id})

@app.exception_handler(RequestValidationError)         # override the default 422 shape
async def validation_handler(request, exc):
    return JSONResponse(422, {"code": "validation_error", "errors": exc.errors()})

Raise HTTPException for genuine HTTP concerns and domain exceptions for business rules, mapping them centrally. Never let a raw exception reach the client — add middleware that catches everything, logs with the request id, and returns a generic 500.

How do you test a FastAPI app?

@pytest.fixture
async def client(app):
    app.dependency_overrides[get_db] = lambda: test_session   # DI makes this trivial
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
        yield c
    app.dependency_overrides.clear()

async def test_create_user(client):
    r = await client.post("/users", json={"email": "a@b.com", "password": "correct-horse-x"})
    assert r.status_code == 201 and "password" not in r.json()

dependency_overrides is the killer feature: swap the database, the clock, or an external API client with no monkeypatching. Use a real database in a container (testcontainers) with a transaction rolled back per test rather than mocking the ORM — mocked ORMs test your mocks.

How do you deploy FastAPI, and how many workers?

uvicorn behind a process manager — gunicorn -k uvicorn.workers.UvicornWorker -w N, or uvicorn --workers N, in a container behind a load balancer. Workers ≈ CPU cores for async workloads (each worker has its own event loop; more workers do not help an I/O-bound app the way they do a sync one). Set --proxy-headers and --forwarded-allow-ips behind a proxy, expose /health and /ready distinctly (liveness vs dependency readiness), and handle SIGTERM with a grace period so in-flight requests finish. Add prometheus-fastapi-instrumentator or OpenTelemetry for metrics and traces, and disable /docs in production if the API is not public.

When would you choose Django over FastAPI, and vice versa?

FastAPI: API-only services, async-heavy I/O (many third-party calls, WebSockets, streaming), teams that want typed contracts and generated OpenAPI clients, small services where you want to choose each component.

Django: anything needing an admin, sessions, templates, a mature ORM with migrations, batteries-included auth and permissions, or a large team benefiting from convention. Django's ORM and admin are worth more than the async gap for most CRUD products, and Django supports ASGI too.

The honest framing: choose FastAPI for services, Django for products. Say "it depends on whether I need Django's ecosystem" rather than declaring a winner.