pkglog
KO

FastAPI in Production: Pydantic v2, async vs sync Endpoints, Dependency Injection and JWT

By · · 34 min read · Updated Intermediate tutorial

Key takeaways

FastAPI is a modern Python web framework for building APIs on ASGI. Its documentation describes performance as on par with Node.js and Go thanks to Starlette and Pydantic; in practice the gain comes from async I/O on I/O-bound endpoints. It ships automatic OpenAPI docs and Pydantic type safety.

Why FastAPI?

FastAPI is a popular Python framework for building APIs. If you’re migrating from Flask, you’ll gain:

  • Higher concurrency for I/O-bound endpoints (async I/O on ASGI: while one request waits on the database or another API, the event loop serves others)
  • Automatic API docs (Swagger UI + ReDoc, zero config)
  • Type safety via Pydantic — catch bugs at startup, not runtime
  • Built-in validation — no more manual request parsing

The throughput difference is mostly about waiting, not raw speed. A sync WSGI worker is tied up for the whole request, including the time it spends waiting on I/O, so concurrency is capped by the number of workers and threads. An async ASGI app can keep many requests in flight on one worker as long as the handlers await non-blocking drivers. For CPU-bound endpoints the frameworks perform much closer to each other, and a blocking call inside async def can make FastAPI slower (see below). Load-test your own endpoints with a tool like wrk or locust before counting on a specific gain.

Installation

pip install fastapi uvicorn[standard]

Your First API (30 seconds)

The q: str | None = None parameter on get_item is doing more work than it looks — FastAPI reads the type hint itself to decide this is an optional query parameter (?q=...), validates it if present, and documents it in Swagger automatically. There’s no separate validation step to write; the function signature is the API contract.

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def root():
    return {"message": "Hello, FastAPI!"}

@app.get("/items/{item_id}")
async def get_item(item_id: int, q: str | None = None):
    return {"item_id": item_id, "q": q}
uvicorn main:app --reload
# → http://localhost:8000
# → http://localhost:8000/docs  (Swagger UI)
# → http://localhost:8000/redoc (ReDoc)

Pydantic Models — Request & Response Validation

Pydantic is FastAPI’s core for data validation and serialization. The split between UserCreate and UserResponse below isn’t boilerplate for its own sake — it’s the same “never reuse your internal model as your public API contract” principle that matters in any typed API framework. UserCreate describes what a client is allowed to send (no id, no created_at — the server owns those), and UserResponse describes what the server promises to return. Collapsing them into one model tends to either leak fields you didn’t mean to expose or force awkward Optional fields on things that should always be required on one side of the request/response boundary.

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field
from datetime import datetime

app = FastAPI()

class UserCreate(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    bio: str | None = None

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

    class Config:
        from_attributes = True  # ORM mode

@app.post("/users/", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    # Validation is automatic — invalid data returns 422
    # Save to DB here...
    return UserResponse(
        id=1,
        name=user.name,
        email=user.email,
        created_at=datetime.now()
    )

What Pydantic does automatically:

  • Rejects invalid data with clear error messages (422 response)
  • Converts types (e.g., "42" → 42 for int fields)
  • Documents the schema in Swagger UI

Routing & HTTP Methods

Path operations map HTTP verbs onto functions the same way most web frameworks do, but the interesting detail here is ItemUpdate.dict(exclude_unset=True) — exclude_unset is what makes PATCH (partial update) actually partial. Without it, every field the client didn’t send would come through as its default (None), and the update would overwrite price and in_stock with None even when the client only meant to change name. exclude_unset returns only the fields the client explicitly included in the request body, which is exactly the set of fields a PATCH should touch.

from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel

app = FastAPI()

# In-memory store for this example
items: dict[int, dict] = {}
next_id = 1

class Item(BaseModel):
    name: str
    price: float
    in_stock: bool = True

class ItemUpdate(BaseModel):
    name: str | None = None
    price: float | None = None
    in_stock: bool | None = None

@app.get("/items/", response_model=list[dict])
async def list_items(skip: int = 0, limit: int = 10):
    return list(items.values())[skip : skip + limit]

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return items[item_id]

@app.post("/items/", status_code=status.HTTP_201_CREATED)
async def create_item(item: Item):
    global next_id
    # .model_dump() is the Pydantic v2 method — .dict() still works but
    # is deprecated and will be removed in a future Pydantic major version
    item_dict = {"id": next_id, **item.model_dump()}
    items[next_id] = item_dict
    next_id += 1
    return item_dict

@app.patch("/items/{item_id}")
async def update_item(item_id: int, item: ItemUpdate):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    update_data = item.model_dump(exclude_unset=True)
    items[item_id].update(update_data)
    return items[item_id]

@app.delete("/items/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    del items[item_id]

Async/Await — When to Use It

The distinction the comments below are pointing at is easy to state but easy to get backwards in practice: async def only helps when the function actually awaits something (a network call, an async DB driver) — it doesn’t make CPU-bound code faster, and using a blocking library (a sync DB driver, requests instead of httpx) inside an async def function is actively worse than a plain def, because it blocks FastAPI’s single event loop for every concurrent request instead of just the one thread handling that request. If you’re not sure whether your dependency has an async version, a plain def (which FastAPI runs in a thread pool automatically) is the safer default than guessing wrong with async def.

import asyncio
import httpx
from fastapi import FastAPI

app = FastAPI()

# ✅ Use async for I/O-bound operations
@app.get("/weather/{city}")
async def get_weather(city: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "your-key"}
        )
        return response.json()

# ✅ Async DB query (using asyncpg, SQLAlchemy async, etc.)
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    # await db.fetch_one(query, values={"id": user_id})
    return {"id": user_id}

# ✅ Regular sync function — also works fine
@app.get("/compute")
def heavy_computation(n: int):
    # CPU-bound: FastAPI runs this in a thread pool automatically
    result = sum(range(n))
    return {"result": result}

Rule of thumb:

  • Database queries, HTTP calls, file I/O → async def
  • CPU-heavy computation → regular def (FastAPI uses a thread pool)

Dependency Injection

FastAPI’s Depends() system makes it easy to share logic (auth, DB sessions, config) across endpoints. The payoff that doesn’t show up until you write tests: because get_current_user is a plain function FastAPI calls for you, app.dependency_overrides[get_current_user] = lambda: {"id": 1, "name": "test-user"} lets a test suite swap in a fake user without touching a real Authorization header or a real token — the endpoint code never needs to know it’s being tested. Inlining the auth check directly into every endpoint instead would mean every test has to construct a real (or realistically-faked) bearer token.

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

app = FastAPI()

# --- Dependency: get current user from token ---
def get_current_user(authorization: str | None = Header(None)):
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Not authenticated")
    token = authorization[7:]
    # In production: validate JWT here
    return {"id": 1, "name": "Alice", "token": token}

CurrentUser = Annotated[dict, Depends(get_current_user)]

# --- Dependency: pagination params ---
def get_pagination(page: int = 1, size: int = 20) -> dict:
    if size > 100:
        raise HTTPException(status_code=400, detail="Max page size is 100")
    return {"skip": (page - 1) * size, "limit": size}

Pagination = Annotated[dict, Depends(get_pagination)]

# --- Using dependencies ---
@app.get("/me")
async def get_me(current_user: CurrentUser):
    return current_user

@app.get("/posts")
async def list_posts(current_user: CurrentUser, pagination: Pagination):
    # Only authenticated users can access this
    return {
        "user": current_user["name"],
        "pagination": pagination,
        "posts": []
    }

Database Integration — SQLAlchemy Async

get_db uses a generator (yield instead of return) specifically so FastAPI can run cleanup code after the request finishes — everything before yield runs before the endpoint, and everything after runs once the response has been sent, regardless of whether the endpoint succeeded or raised. That’s what makes the commit-on-success / rollback-on-exception pattern here work correctly: the try/except wraps the yield, so an exception raised inside the endpoint still propagates back through this generator, triggering the rollback, before FastAPI re-raises it to its own exception handling. Forgetting the try/except here is a common way to end up with a connection pool slowly leaking sessions that were never properly closed after a failed request.

from fastapi import FastAPI, Depends
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy import Column, Integer, String, DateTime, select
from sqlalchemy.orm import DeclarativeBase
from datetime import datetime

# Database setup
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/dbname"
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(255), unique=True, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

# DB session dependency
async def get_db():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

app = FastAPI()

@app.get("/users/{user_id}")
async def get_user(user_id: int, db: AsyncSession = Depends(get_db)):
    result = await db.execute(select(User).where(User.id == user_id))
    user = result.scalar_one_or_none()
    if not user:
        from fastapi import HTTPException
        raise HTTPException(status_code=404, detail="User not found")
    return {"id": user.id, "name": user.name, "email": user.email}

class UserCreatePayload(BaseModel):
    name: str
    email: str

# ⚠️ Bare `name: str, email: str` parameters (no Pydantic model, no
# Body()) are treated as QUERY parameters by FastAPI, not JSON body
# fields — so this endpoint would actually expect
# POST /users/?name=Alice&[email protected], not a JSON body.
# For a resource-creation endpoint, a request model like the one
# above is what you almost always want instead:
@app.post("/users/")
async def create_user(payload: UserCreatePayload, db: AsyncSession = Depends(get_db)):
    user = User(name=payload.name, email=payload.email)
    db.add(user)
    await db.flush()  # Get the auto-generated ID
    return {"id": user.id, "name": user.name}

JWT Authentication

The 30-minute access token here is a reasonable default for a first pass, but it’s worth knowing this pattern’s real limitation before relying on it in production: a plain JWT like this can’t be revoked before it expires. If a token leaks, or a user needs to be logged out immediately (a password reset, a compromised account), there’s no way to invalidate it short of rotating SECRET_KEY for everyone. Production systems typically either keep access tokens very short-lived (5-15 minutes) paired with a revocable refresh token, or maintain a denylist of revoked token IDs (jti claim) checked on every request — this example skips both to keep the auth flow readable.

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

SECRET_KEY = "your-secret-key-min-32-chars"  # ⚠️ example only — load
# this from an environment variable in real code (e.g. os.environ["JWT_SECRET"]).
# A hardcoded secret committed to source control lets anyone who can
# read the repo forge valid tokens for any user.
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

app = FastAPI()

def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
        if user_id is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    # In production: fetch user from DB here
    return {"id": int(user_id), "name": "Alice"}

@app.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends()):
    # In production: look up user in DB and verify password
    # user = await get_user(form_data.username)
    # if not verify_password(form_data.password, user.hashed_password):
    #     raise HTTPException(...)
    
    access_token = create_access_token(
        data={"sub": "1"},  # user ID
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    return {"access_token": access_token, "token_type": "bearer"}

@app.get("/me")
async def read_me(current_user: dict = Depends(get_current_user)):
    return current_user

Background Tasks

BackgroundTasks runs send_welcome_email after the response has already been sent to the client, in the same process — it’s not a durable job queue. That distinction matters the first time a deploy happens to restart the server between the response being sent and the background task actually running: the task is simply gone, silently, with no retry and no record it was ever supposed to run. Fire-and-forget notifications (an email, a log line) are a reasonable fit; anything that must not be silently dropped (a payment webhook, a critical audit log) belongs in a real task queue like Celery or RQ instead.

from fastapi import FastAPI, BackgroundTasks
import smtplib

app = FastAPI()

def send_welcome_email(email: str, name: str):
    # Runs after the response is sent
    print(f"Sending welcome email to {email}...")
    # smtplib.SMTP(...).sendmail(...)

@app.post("/register")
async def register(
    email: str, 
    name: str, 
    background_tasks: BackgroundTasks
):
    # Immediately return a response
    background_tasks.add_task(send_welcome_email, email, name)
    return {"message": "Registration successful. Check your email."}

Error Handling

Custom exception handlers like app_error_handler exist so that domain logic can raise a meaningful exception (AppError("Something went wrong", "OPERATION_FAILED")) instead of constructing an HTTP response by hand deep inside a service function that shouldn’t know or care about HTTP status codes. The RequestValidationError override below is worth keeping in most real projects — FastAPI’s default 422 body is accurate but verbose and inconsistent with a typical API’s error shape; reshaping it into a flat {"field": ..., "message": ...} list up front means every client-facing error, validation or otherwise, comes back in the same predictable format.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# Custom exception
class AppError(Exception):
    def __init__(self, message: str, code: str = "ERROR"):
        self.message = message
        self.code = code

@app.exception_handler(AppError)
async def app_error_handler(request: Request, exc: AppError):
    return JSONResponse(
        status_code=400,
        content={"error": exc.code, "message": exc.message}
    )

# Override validation error format
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    errors = []
    for error in exc.errors():
        errors.append({
            "field": ".".join(str(x) for x in error["loc"]),
            "message": error["msg"],
            "type": error["type"]
        })
    return JSONResponse(status_code=422, content={"errors": errors})

@app.get("/risky")
async def risky_endpoint(fail: bool = False):
    if fail:
        raise AppError("Something went wrong", "OPERATION_FAILED")
    return {"status": "ok"}

Project Structure (Production)

myapi/
├── main.py              # App entry point
├── core/
│   ├── config.py        # Settings (pydantic-settings)
│   └── security.py      # JWT, password hashing
├── db/
│   ├── database.py      # Engine, session factory
│   └── models.py        # SQLAlchemy models
├── api/
│   ├── deps.py          # Shared dependencies
│   └── routes/
│       ├── users.py     # /users endpoints
│       ├── items.py     # /items endpoints
│       └── auth.py      # /token endpoint
├── schemas/
│   ├── user.py          # UserCreate, UserResponse, etc.
│   └── item.py
└── tests/
    ├── test_users.py
    └── conftest.py
# main.py — Router registration
from fastapi import FastAPI
from api.routes import users, items, auth

app = FastAPI(title="My API", version="1.0.0")

app.include_router(auth.router, prefix="/auth", tags=["auth"])
app.include_router(users.router, prefix="/users", tags=["users"])
app.include_router(items.router, prefix="/items", tags=["items"])

Deployment

Development

uvicorn main:app --reload --host 0.0.0.0 --port 8000

Production

Running Uvicorn directly (uvicorn main:app) is fine for development, but a single Uvicorn process is a single point of failure and can only use one CPU core. Gunicorn in front of Uvicorn workers (-w 4 here spawns 4 worker processes) gives you process-level parallelism and, more importantly, automatic worker restarts if one crashes — a request that segfaults a worker doesn’t take the whole service down with it, Gunicorn just spins up a replacement. I’ve seen teams skip this step because “it works fine with plain Uvicorn in staging,” only to find out in production that a single unhandled exception in a C extension (a bad numpy call, a native DB driver bug) can kill the entire process instead of just failing one request — which Gunicorn’s worker isolation would have contained.

# Gunicorn + Uvicorn workers
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

Docker

Configuration file:

FROM python:3.12-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .

CMD ["gunicorn", "main:app", "-w", "4", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8000"]