FastAPI Interview Questions

Summary

FastAPI is a modern ASGI Python web framework built on Starlette and Pydantic. It uses type hints to drive request validation, serialization, dependency injection and automatic OpenAPI docs.

Longer Answer

Where Flask is a minimal WSGI microframework and Django is a batteries-included full-stack framework, FastAPI focuses on APIs. Declaring def get_item(item_id: int, q: str | None = None) is enough for FastAPI to parse, validate and document the parameters. It supports async def natively, generates Swagger UI (/docs) and ReDoc (/redoc), and is among the fastest Python frameworks thanks to Starlette and Uvicorn.

Code Example

from fastapi import FastAPI

app = FastAPI(title="Inventory API")

@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str | None = None) -> dict:
    return {"item_id": item_id, "q": q}

# run: uvicorn main:app --reload

Key Points

• ASGI, built on Starlette + Pydantic
• Type hints → validation, docs and editor support
• Native async support
• Automatic OpenAPI/Swagger docs
• Dependency injection system via Depends

Comparison / Tradeoffs

Summary

Parameters named in the path template are path parameters; simple-typed parameters not in the path are query parameters; parameters typed as Pydantic models are read from the JSON body.

Longer Answer

You can make the source explicit with Path(), Query(), Body(), Header(), Cookie() and Form(), usually via Annotated for metadata like validation constraints and descriptions. Optional query params have defaults (q: str | None = None). Lists from repeated query keys need Query(): tags: Annotated[list[str], Query()] = [].

Code Example

from typing import Annotated
from fastapi import FastAPI, Path, Query
from pydantic import BaseModel

app = FastAPI()

class ItemUpdate(BaseModel):
    name: str
    price: float

@app.put("/items/{item_id}")
async def update_item(
    item_id: Annotated[int, Path(ge=1)],
    item: ItemUpdate,
    notify: Annotated[bool, Query()] = False,
):
    return {"id": item_id, **item.model_dump(), "notify": notify}

Key Points

• Path template names → path params
• Scalars otherwise → query params
• Pydantic models → body
• Annotated[int, Query(ge=1)] adds constraints

Summary

Request bodies, query models and settings are declared as Pydantic BaseModel classes. FastAPI parses incoming data into them, validates types and constraints, and returns a 422 response with detailed errors on failure.

Longer Answer

Pydantic v2 (with its Rust core, pydantic-core) coerces compatible input (e.g., '3' → 3 in lax mode), enforces constraints (Field(gt=0, max_length=50)), and supports custom validators (@field_validator, @model_validator). Errors include the location (body.price), message and type, which clients can map to form fields.

Code Example

from pydantic import BaseModel, Field, field_validator, EmailStr

class SignUp(BaseModel):
    email: EmailStr
    password: str = Field(min_length=12)
    age: int = Field(ge=18)

    @field_validator("password")
    @classmethod
    def has_digit(cls, v: str) -> str:
        if not any(c.isdigit() for c in v):
            raise ValueError("password must contain a digit")
        return v

Key Points

• Models declared with type hints and Field() constraints
• Invalid input → automatic 422 response
• @field_validator / @model_validator for custom rules
• Strict mode disables coercion

Summary

It declares the output schema: FastAPI validates and filters the returned data through that model, so extra fields (like password hashes) are excluded, and it documents the response in OpenAPI.

Longer Answer

You can use a return type annotation (-> UserOut) or response_model=UserOut (needed when the function returns something else, like an ORM object or dict). Options like response_model_exclude_unset=True omit fields not explicitly set. Separate input and output models (UserCreate with a password, UserOut without) to avoid leaking data.

Code Example

class UserCreate(BaseModel):
    email: EmailStr
    password: str

class UserOut(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    email: EmailStr

@app.post("/users", response_model=UserOut, status_code=201)
async def create_user(data: UserCreate, db: DbSession):
    user = await users.create(db, data)  # ORM object with hashed_password
    return user  # hashed_password is filtered out

Key Points

• Filters output to the declared schema
• Documents responses in OpenAPI
• Separate In/Out models
• Return ORM objects with from_attributes=True

Summary

You declare a parameter as Annotated[T, Depends(provider)]; FastAPI calls the provider per request (resolving its own dependencies recursively), caches the result within that request, and injects it.

Longer Answer

Dependencies can be functions, async functions, classes or generators (yield dependencies for setup/teardown like DB sessions). They can declare their own query params, headers or other dependencies, which appear in the OpenAPI docs. Override them in tests with app.dependency_overrides[provider] = fake. Router- or app-level dependencies=[Depends(verify_token)] run for all routes without injecting a value.

Code Example

from typing import Annotated
from fastapi import Depends, Header, HTTPException

async def get_api_key(x_api_key: Annotated[str, Header()]) -> str:
    if x_api_key != settings.api_key:
        raise HTTPException(status_code=401, detail="Invalid API key")
    return x_api_key

ApiKey = Annotated[str, Depends(get_api_key)]

@app.get("/reports")
async def reports(_: ApiKey):
    return {"ok": True}

Key Points

• Declarative, per-request resolution
• Sub-dependencies resolved recursively
• Cached per request by default (use_cache=False to disable)
• dependency_overrides for testing

Summary

A dependency that yields provides the yielded value to the endpoint and runs the code after yield once the request is done, which suits opening and closing DB sessions or transactions.

Longer Answer

FastAPI wraps the generator like a context manager. Exceptions raised in the endpoint propagate into the dependency at the yield, so you can roll back in except. Exactly when teardown runs relative to sending the response has changed between FastAPI versions, so commit explicitly in your service/endpoint (so commit errors reach the client) and let background tasks open their own sessions instead of reusing the request's.

Code Example

from collections.abc import AsyncIterator
from sqlalchemy.ext.asyncio import AsyncSession

async def get_db() -> AsyncIterator[AsyncSession]:
    async with SessionLocal() as session:
        try:
            yield session  # endpoint/service calls `await session.commit()`
        except Exception:
            await session.rollback()
            raise

DbSession = Annotated[AsyncSession, Depends(get_db)]

Key Points

• Code before yield = setup
• Code after yield = teardown
• Catch exceptions to roll back
• Background tasks should open their own sessions

Summary

Use async def when the endpoint only awaits async I/O (async DB drivers, httpx). Use plain def when it calls blocking libraries, because FastAPI runs def endpoints in a threadpool so they don't block the event loop.

Longer Answer

The danger is calling blocking code (e.g., requests.get, time.sleep, sync SQLAlchemy, CPU-heavy work) inside async def. That freezes the event loop for every concurrent request. Options: switch to async libraries, make the endpoint def, or offload with await run_in_threadpool(func) / anyio.to_thread.run_sync. CPU-bound work belongs in a process pool or a task queue.

Code Example

import httpx
from fastapi.concurrency import run_in_threadpool

@app.get("/weather")
async def weather(city: str):
    async with httpx.AsyncClient(timeout=5) as client:  # non-blocking
        r = await client.get(f"https://api.example.com/w?city={city}")
    return r.json()

@app.get("/legacy")
def legacy_report():  # sync def: runs in threadpool
    return blocking_sdk.generate()

@app.get("/mixed")
async def mixed():
    data = await run_in_threadpool(blocking_sdk.generate)
    return data

Key Points

• async def + only non-blocking awaits
• def endpoints run in a threadpool (default ~40 threads)
• Never block inside async def
• CPU-bound → process pool / Celery

Summary

Use OAuth2PasswordBearer(tokenUrl="token") as a dependency that extracts the Bearer token, issue signed JWTs (PyJWT) at the token endpoint after verifying the password hash, and decode/verify the token in a get_current_user dependency.

Longer Answer

Hash passwords with pwdlib/passlib (Argon2 or bcrypt). The token includes sub (user ID) and exp. get_current_user decodes with a pinned algorithm, loads the user, and raises 401 with a WWW-Authenticate: Bearer header on failure. Layer authorization with dependencies like require_role('admin') or OAuth2 scopes via Security().

Code Example

import jwt
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.post("/token")
async def login(form: Annotated[OAuth2PasswordRequestForm, Depends()], db: DbSession):
    user = await authenticate(db, form.username, form.password)
    if not user:
        raise HTTPException(401, "Incorrect credentials", headers={"WWW-Authenticate": "Bearer"})
    token = jwt.encode({"sub": str(user.id), "exp": utcnow() + timedelta(minutes=30)}, SECRET, algorithm="HS256")
    return {"access_token": token, "token_type": "bearer"}

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

Key Points

• OAuth2PasswordBearer extracts the token
• Sign JWTs with expiry; pin algorithms
• get_current_user dependency
• Role/scope checks as composable dependencies

Summary

Create an async engine (create_async_engine with asyncpg), an async_sessionmaker, provide a session per request via a yield dependency, and use select() with await session.execute() or session.scalars().

Longer Answer

Set expire_on_commit=False so returned objects stay usable after commit. Lazy loading doesn't work implicitly in async (it raises MissingGreenlet), so eager-load relationships with selectinload()/joinedload(). Configure pool size based on worker count and the DB's max connections. Use Alembic for migrations.

Code Example

from sqlalchemy import select
from sqlalchemy.orm import selectinload
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker

engine = create_async_engine(settings.database_url, pool_size=10, max_overflow=5)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@router.get("/{user_id}/orders")
async def user_orders(user_id: int, db: DbSession):
    stmt = select(User).options(selectinload(User.orders)).where(User.id == user_id)
    user = (await db.scalars(stmt)).one_or_none()
    if user is None:
        raise HTTPException(404)
    return user.orders

Key Points

• create_async_engine('postgresql+asyncpg://...')
• Session per request via yield dependency
• expire_on_commit=False
• Eager loading (selectinload) instead of lazy loads

Summary

The ASGI server calls the app; Starlette's outer ServerErrorMiddleware wraps your middleware (the last added runs first), then ExceptionMiddleware applies your exception handlers; the router matches the path; FastAPI resolves dependencies (once per request by default) and validates parameters and the body with Pydantic (422 on failure); the endpoint runs (async def on the event loop, def in a thread pool); the return value is serialised through the response model; dependency exit code and background tasks run afterwards.

Longer Answer

Knowing the layers explains common surprises: an HTTPException raised in middleware becomes a 500 because it's outside the exception layer; validation errors never reach your endpoint; a blocking call in an async def endpoint stalls every request on the event loop. Exactly when dependencies with yield run their exit code relative to sending the response has changed between FastAPI versions, so check the documentation for your version.

Key Points

• ServerErrorMiddleware → your middleware → ExceptionMiddleware → router.
• Dependencies resolved and cached per request.
• Pydantic validation, 422 on failure.
• async def on the loop, def in a thread pool.

Deep dive: Request Lifecycle course: FastAPI and Starlette: Middleware, Exceptions and Dependencies

Summary

WSGI is the synchronous Python interface between servers and apps (one request per call). ASGI is its async successor, supporting async request handling, WebSockets, HTTP/2 and long-lived connections.

Longer Answer

A WSGI app is a callable app(environ, start_response) that blocks a worker thread per request. An ASGI app is async def app(scope, receive, send), so one event loop can multiplex thousands of concurrent connections. FastAPI/Starlette are ASGI apps served by Uvicorn, Hypercorn or Granian.

Key Points

• WSGI: sync, request/response only
• ASGI: async, WebSockets, lifespan events
• Servers: Uvicorn, Hypercorn, Granian
• ASGI apps can still run sync code in a threadpool

Deep dive: Request Lifecycle course: WSGI, ASGI and the Event Loop

Summary

Use the lifespan parameter with an async context manager: code before yield runs at startup (create DB pools, HTTP clients, load ML models) and code after runs at shutdown.

Longer Answer

The older @app.on_event("startup") handlers are deprecated. Lifespan can yield a state dict, available as request.state (Starlette 0.26+), or store objects on app.state. With multiple Uvicorn workers, lifespan runs once per worker process.

Code Example

from contextlib import asynccontextmanager
import httpx

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.http = httpx.AsyncClient(timeout=10)
    app.state.model = load_model("model.bin")
    yield
    await app.state.http.aclose()

app = FastAPI(lifespan=lifespan)

Key Points

• lifespan= async context manager
• Replaces deprecated on_event
• Runs per worker process
• Good place for shared clients/pools

Summary

BackgroundTasks runs small functions after the response is sent, in the same process; for heavy, slow, retryable or critical jobs, use a queue like Celery, RQ, Dramatiq, arq or a managed queue.

Longer Answer

Background tasks share the server process: if the worker crashes or restarts, tasks are lost; there are no retries, scheduling or visibility; and CPU-heavy tasks steal capacity from requests. They're fine for sending a notification email or writing an audit log.

Code Example

from fastapi import BackgroundTasks

def send_welcome_email(email: str) -> None:
    mailer.send(email, "Welcome!")

@app.post("/signup", status_code=201)
async def signup(data: SignUp, tasks: BackgroundTasks):
    user = await create_user(data)
    tasks.add_task(send_welcome_email, user.email)
    return {"id": user.id}

Key Points

• Runs after response, same process
• No persistence or retries
• Good for small fire-and-forget work
• Use Celery/arq/RQ for durable jobs

Summary

Add middleware with app.add_middleware(...) (e.g., CORSMiddleware, GZipMiddleware, TrustedHostMiddleware) or write HTTP middleware with @app.middleware("http").

Longer Answer

CORS: list exact allow_origins (avoid * with allow_credentials=True; browsers reject it). Middleware wraps every request, so it suits timing, request IDs and security headers. For performance-critical code, pure ASGI middleware avoids the overhead of BaseHTTPMiddleware. Middleware order matters: the last added is the outermost.

Code Example

import time, uuid
from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

@app.middleware("http")
async def add_request_id(request, call_next):
    request_id = request.headers.get("x-request-id", str(uuid.uuid4()))
    start = time.perf_counter()
    response = await call_next(request)
    response.headers["x-request-id"] = request_id
    response.headers["server-timing"] = f"app;dur={(time.perf_counter() - start) * 1000:.1f}"
    return response

Key Points

• CORSMiddleware with explicit origins
• @app.middleware('http') for simple cases
• Pure ASGI middleware for performance
• Order: last added runs first

Deep dive: Request Lifecycle course: FastAPI and Starlette: Middleware, Exceptions and Dependencies

Summary

Raise HTTPException(status_code, detail) for expected errors, define domain exceptions with @app.exception_handler(MyError) to map them to responses, and override RequestValidationError handling to customize 422 payloads.

Longer Answer

Keep domain code framework-agnostic by raising your own exceptions (OrderNotFound) and translating them at the API boundary. A catch-all handler for Exception should log with a request ID and return a generic 500 without internals. Consider RFC 9457 Problem Details as a consistent error format.

Code Example

class OrderNotFound(Exception):
    def __init__(self, order_id: int):
        self.order_id = order_id

@app.exception_handler(OrderNotFound)
async def order_not_found(request, exc: OrderNotFound):
    return JSONResponse(status_code=404, content={"title": "Order not found", "order_id": exc.order_id})

Key Points

• HTTPException for HTTP-level errors
• Exception handlers map domain errors
• Custom RequestValidationError handler
• Never leak stack traces

Summary

Split routes into APIRouter modules per domain (users, orders), included with prefixes and tags, and separate schemas (Pydantic), models (ORM), services/repositories and dependencies into their own modules.

Longer Answer

A common layout: app/main.py (create app, include routers), app/core/config.py (settings), app/db/ (engine, session), app/<domain>/router.py, schemas.py, service.py, models.py. Routers can declare shared dependencies and responses. Keep business logic out of route functions so it's testable and reusable by CLI/workers.

Code Example

# app/users/router.py
router = APIRouter(prefix="/users", tags=["users"])

@router.get("/{user_id}", response_model=UserOut)
async def get_user(user_id: int, db: DbSession):
    return await service.get_user(db, user_id)

# app/main.py
app.include_router(users.router, prefix="/api/v1")

Key Points

• APIRouter(prefix='/users', tags=['users'])
• Domain-oriented packages
• Thin routes → services
• Centralized settings and DB session dependency

Summary

Use TestClient (sync, based on httpx) or httpx.AsyncClient with ASGITransport for async tests, override dependencies (DB, auth, settings) with app.dependency_overrides, and use pytest fixtures for setup.

Longer Answer

For DB tests, run each test in a transaction that is rolled back, or use a disposable test database (e.g., Testcontainers). Wrap TestClient(app) in a with block to trigger lifespan events. Test both happy paths and validation errors (422), and use pytest-asyncio or anyio for async tests.

Code Example

from fastapi.testclient import TestClient

def fake_user():
    return User(id=1, email="[email protected]", role="admin")

def test_reports_requires_admin():
    app.dependency_overrides[get_current_user] = fake_user
    with TestClient(app) as client:
        r = client.get("/reports")
    assert r.status_code == 200
    app.dependency_overrides.clear()

Key Points

• TestClient(app) in a with block runs lifespan
• dependency_overrides to swap DB/auth
• httpx.AsyncClient(transport=ASGITransport(app)) for async
• Isolated DB per test

Summary

Run Uvicorn with multiple worker processes (uvicorn --workers N or Gunicorn with uvicorn.workers.UvicornWorker), behind a reverse proxy or load balancer, usually in containers, with health checks and proper proxy header handling.

Longer Answer

Rule of thumb: start with workers ≈ CPU cores and tune with load tests. In Kubernetes, many teams run one worker per container and scale pods instead. Set --proxy-headers and --forwarded-allow-ips behind a proxy so request.client and scheme are right. The FastAPI CLI (fastapi run) wraps Uvicorn for production.

Code Example

# Dockerfile (simplified)
FROM python:3.13-slim
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev
COPY app ./app
CMD ["uv", "run", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4", "--proxy-headers"]

Key Points

• Multiple worker processes for CPU parallelism
• Reverse proxy/LB in front (TLS, buffering)
• One worker per container in Kubernetes is common
• Health/readiness endpoints

Summary

Profile first, then: use async drivers end to end, avoid blocking calls in async def, reuse HTTP clients and DB pools, cache hot reads (Redis), offload heavy work to queues, optimize serialization, and scale workers horizontally.

Longer Answer

Common hotspots are DB queries (indexes, N+1), per-request client creation (create one httpx.AsyncClient in lifespan), and large Pydantic payloads (use response_model thoughtfully, ORJSON, or return pre-serialized data). Tune the DB pool against worker count so you don't exhaust Postgres connections (use PgBouncer). Load test with Locust/k6 and watch event loop lag.

Key Points

• No blocking calls in async paths
• Shared clients and pools via lifespan
• Cache + queue offloading
• Pool sizing vs DB connection limits
• Load test and measure event loop lag

Summary

A coroutine (async def) is a pausable function; the event loop runs coroutines, switching between them whenever one awaits I/O; a Task wraps a coroutine so it runs concurrently. FastAPI runs each async def request handler as a coroutine on Uvicorn's event loop.

Longer Answer

Concurrency comes from cooperative yielding at await points, so a single thread can serve thousands of waiting requests. If a coroutine never awaits (CPU work) or calls blocking functions, nothing else can run. asyncio.create_task() schedules work concurrently, but tasks started inside a request without tracking can be garbage-collected or outlive the request.

Key Points

• Coroutines yield control at await
• Single-threaded cooperative scheduling
• Tasks run coroutines concurrently
• Blocking code stalls the whole loop

Summary

Yes: use plain def endpoints and a sync session dependency, and FastAPI runs them in a threadpool. It's simpler and mature, but concurrency is limited by the threadpool size (about 40 by default via AnyIO) and the DB pool.

Longer Answer

Never call a sync session from async def, because it blocks the loop. Mixing styles is fine per endpoint. You can raise the AnyIO thread limiter if needed. For high-concurrency, I/O-heavy services, async drivers scale better.

Key Points

• Sync DB → def endpoints
• Threadpool capacity limits concurrency
• Never use sync sessions in async def