Skip to content

FastAPI Folha de referência

Modern, fast Python web framework for building APIs.

01

Getting Started

First API

FastAPI is built on Starlette and Pydantic. uvicorn is the ASGI server. --reload enables auto-reload during development.

fastapi
# install FastAPI and uvicorn
pip install fastapi uvicorn

# main.py
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

# run: uvicorn main:app --reload

Run with Uvicorn

The app instance 'main:app' means module 'main', variable 'app'. Use --reload only in development; in production run without it and use a process manager.

fastapi
# development with auto-reload
uvicorn main:app --reload

# specify host and port
uvicorn main:app --host 0.0.0.0 --port 8000

# run programmatically
import uvicorn
if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=8000, reload=True)

Async Path Operations

You can declare path operations with either 'def' (run in threadpool) or 'async def'. Use async def when the function performs I/O via async libraries; mixing blocking calls in async def will block the event loop.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
async def read_root():
    return {"message": "hello"}

@app.get("/sync")
def read_sync():
    return {"message": "sync also works"}

# use async when calling async libs (httpx, databases)
@app.get("/data")
async def fetch_data():
    import httpx
    async with httpx.AsyncClient() as client:
        r = await client.get("https://api.example.com")
    return r.json()

Path Operation Decorators

FastAPI supports all HTTP methods: GET, POST, PUT, PATCH, DELETE, OPTIONS, HEAD. Each maps to a decorator. Use status_code= on the decorator to override the default success status.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/items")
def list_items(): ...

@app.post("/items")
def create_item(): ...

@app.put("/items/{item_id}")
def replace_item(item_id: int): ...

@app.patch("/items/{item_id}")
def update_item(item_id: int): ...

@app.delete("/items/{item_id}")
def delete_item(item_id: int): ...

@app.options("/items")
def options_items(): ...

@app.head("/items")
def head_items(): ...

JSON Response & Status

Returning a dict/list is auto-converted to JSON with 200 (or your status_code). For full control over status, headers, or content, return a JSONResponse or Response directly.

fastapi
from fastapi import FastAPI, status
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/plain")
def plain():
    return {"key": "value"}  # auto JSON, 200 OK

@app.post("/created", status_code=status.HTTP_201_CREATED)
def created():
    return {"id": 1}

@app.get("/custom")
def custom():
    return JSONResponse(
        status_code=418,
        content={"error": "I'm a teapot"},
        headers={"X-Custom": "yes"},
    )

Project Structure

Split a growing app into modules with APIRouter, then register them via app.include_router(). Keep Pydantic schemas, DB setup, and dependencies in separate files for maintainability.

fastapi
myapp/
  main.py              # creates the FastAPI app
  routers/
    users.py           # APIRouter for /users
    items.py           # APIRouter for /items
  models/
    schemas.py         # Pydantic models
    database.py        # DB engine & session
  dependencies.py      # shared Depends functions
  tests/
    test_main.py

# main.py
from fastapi import FastAPI
from routers import users, items
app = FastAPI()
app.include_router(users.router)
app.include_router(items.router, prefix="/items")
02

Path Parameters

Basic Path Parameters

Values captured from the path are strings unless you add a type annotation. The parameter name in the path must match the function argument name exactly.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id):
    return {"item_id": item_id}

# GET /items/42 -> {"item_id": "42"}  (string by default)

Type Conversion

Adding a type annotation (int, float, bool, str, Enum) makes FastAPI validate and convert the path value. Invalid values return a 422 response with a clear error message.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):
    return {"item_id": item_id}

# GET /items/3   -> {"item_id": 3}
# GET /items/abc -> 422 Validation Error (not an integer)

Path Validation

Path() adds metadata and constraints to a path parameter. Numeric constraints: ge (>=), gt (>), le (<=), lt (<). Path parameters are always required, so Path() cannot set a default.

fastapi
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(
    item_id: int = Path(
        title="The ID of the item",
        description="Must be a positive integer",
        ge=1,
        le=1000,
    ),
):
    return {"item_id": item_id}

Numeric Constraints

Use ge/gt/le/lt on int or float path parameters to enforce numeric ranges. Combined constraints give you precise bounds validation with automatic 422 errors.

fastapi
from fastapi import FastAPI, Path

app = FastAPI()

@app.get("/ratio/{value}")
def ratio(
    value: float = Path(gt=0, lt=1),  # 0 < value < 1
):
    return {"value": value}

@app.get("/page/{page}")
def page(
    page: int = Path(ge=1),       # >= 1
    size: int = Path(le=100),     # <= 100
):
    return {"page": page, "size": size}

Route Order Matters

Routes are matched in declaration order. Declare specific paths like /users/me before dynamic patterns like /users/{user_id}, or the dynamic route will shadow the specific one.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/users/me")
def read_current_user():
    return {"user": "me"}

@app.get("/users/{user_id}")
def read_user(user_id: str):
    return {"user_id": user_id}

# /users/me MUST be declared before /users/{user_id}
# otherwise 'me' would match the {user_id} pattern

Enum Path Parameters

Using an Enum type for a path parameter restricts it to predefined values and generates an enum schema in the OpenAPI docs. Subclass (str, Enum) so values serialize as strings.

fastapi
from enum import Enum
from fastapi import FastAPI

class ModelName(str, Enum):
    alexnet = "alexnet"
    resnet = "resnet"
    lenet = "lenet"

app = FastAPI()

@app.get("/models/{model_name}")
def get_model(model_name: ModelName):
    if model_name is ModelName.alexnet:
        return {"model": model_name, "layers": 5}
    return {"model": model_name}

# GET /models/resnet -> {"model": "resnet"}
03

Query Parameters

Basic Query Parameters

Function parameters that are NOT in the path become query parameters. Providing a default makes them optional; the default is used when the query key is absent.

fastapi
from fastapi import FastAPI

app = FastAPI()

fake_items = [{"item": "a"}, {"item": "b"}, {"item": "c"}]

@app.get("/items")
def list_items(skip: int = 0, limit: int = 10):
    return fake_items[skip : skip + limit]

# GET /items?skip=0&limit=2
# GET /items                -> skip=0, limit=10

Optional Query Parameters

Use Optional[str] = None (or str | None = None on Python 3.10+) for a query parameter that may be omitted. Required parameters simply have no default value.

fastapi
from typing import Optional
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: str, q: Optional[str] = None):
    if q:
        return {"item_id": item_id, "q": q}
    return {"item_id": item_id}

# GET /items/1?q=hello
# GET /items/1            (q is None)

Query Validation

Query() adds validation: min_length/max_length for strings, ge/gt/le/lt for numbers, pattern (regex) for string format. Use default= to set a value while still applying constraints.

fastapi
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/search")
def search(q: str = Query(min_length=3, max_length=50, pattern="^[a-zA-Z]+$")):
    return {"q": q}

@app.get("/page")
def page(limit: int = Query(default=10, ge=1, le=100)):
    return {"limit": limit}

Boolean Conversion

Bool query parameters accept many truthy values: true, 1, yes, on (case-insensitive). Anything else is interpreted as false. Invalid types like ?active=maybe return 422.

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/flags")
def flags(active: bool = False):
    return {"active": active}

# GET /flags?active=true   -> {"active": true}
# GET /flags?active=1      -> {"active": true}
# GET /flags?active=yes    -> {"active": true}
# GET /flags?active=off    -> {"active": false}

List Query Parameters

A List[str] parameter accepts the same key repeated in the query string. Use Query(default=[]) for an empty list by default, or Query(default=None) to allow missing values.

fastapi
from typing import List
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
def list_items(q: List[str] = Query(default=[])):
    return {"q": q}

# GET /items?q=a&q=b&q=c  -> {"q": ["a", "b", "c"]}
# declare default=None to accept a list or null

Required Query Parameters

Use Query(...) (the Ellipsis) to declare a required query parameter that still has validation metadata. Without a default and without Query(...), a plain annotated param is also required.

fastapi
from fastapi import FastAPI, Query

app = FastAPI()

@app.get("/items")
def read_item(q: str = Query(...)):
    return {"q": q}

# GET /items?q=hello   -> ok
# GET /items           -> 422 (q is required)
# Query(...) (Ellipsis) marks a parameter as required
04

Request Body (Pydantic)

BaseModel Basics

A Pydantic BaseModel declared as a function parameter becomes the JSON request body. FastAPI validates the payload, converts types, and returns 422 on invalid data. Fields with defaults are optional.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel

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

app = FastAPI()

@app.post("/items")
def create_item(item: Item):
    return item

# POST /items  body: {"name": "Apple", "price": 0.5}
# -> {"name": "Apple", "price": 0.5, "in_stock": true}

Field Types & Constraints

Field() adds constraints and metadata to model attributes: min_length/max_length for strings, gt/ge/lt/le for numbers, default_factory for mutable defaults. model_config with json_schema_extra adds examples to the docs.

fastapi
from pydantic import BaseModel, Field
from typing import Optional

class Item(BaseModel):
    name: str = Field(min_length=1, max_length=100)
    price: float = Field(gt=0, description="must be positive")
    tax: Optional[float] = None
    tags: list[str] = Field(default_factory=list)

    model_config = {
        "json_schema_extra": {
            "examples": [{"name": "Apple", "price": 0.5, "tags": ["fruit"]}]
        }
    }

Nested Models

Models can be nested: declare another BaseModel as a field type. FastAPI validates the full nested structure recursively, and the generated OpenAPI schema reflects the nested objects.

fastapi
from pydantic import BaseModel
from typing import Optional

class Image(BaseModel):
    url: str
    name: str

class Item(BaseModel):
    name: str
    description: Optional[str] = None
    image: Optional[Image] = None
    images: list[Image] = []

# valid body:
# {"name": "Phone", "image": {"url": "http://x/a.png", "name": "a"}}

Field Validation

Use @field_validator (Pydantic v2) to add custom validation logic per field. Raise ValueError to reject input (FastAPI surfaces it as 422). Return the (possibly transformed) value to keep it.

fastapi
from pydantic import BaseModel, field_validator

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

    @field_validator("email")
    @classmethod
    def email_must_contain_at(cls, v: str) -> str:
        if "@" not in v:
            raise ValueError("must contain @")
        return v.lower()

    @field_validator("name")
    @classmethod
    def name_stripped(cls, v: str) -> str:
        return v.strip()

Optional & Default Fields

A field is required only if it has no default and is not Optional. Optional[str] = None is the standard pattern for an optional nullable field; a plain default like 0.0 makes a field optional with that value.

fastapi
from pydantic import BaseModel
from typing import Optional

class Item(BaseModel):
    name: str                       # required
    description: Optional[str] = None  # optional, defaults to None
    price: float = 0.0              # optional, defaults to 0.0
    tags: list[str] = []            # optional, defaults to empty list

# All of these are valid:
# {"name": "X"}
# {"name": "X", "price": 9.99, "tags": ["a"]}

Body with Path & Query

Path, query, and body parameters can coexist in one operation. FastAPI decides by: path param if in the path, Pydantic model (or Body()) for the body, otherwise query. Optional body needs a default of None.

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

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

app = FastAPI()

@app.put("/items/{item_id}")
def update_item(
    item_id: int = Path(ge=1),
    q: Optional[str] = Query(default=None),
    item: Item = None,
):
    result = {"item_id": item_id, **item.model_dump()}
    if q:
        result["q"] = q
    return result
05

Response Model

response_model Basics

response_model filters the returned data to only the fields defined in the model, even if you return an object with extra fields. This is the standard way to hide sensitive fields like passwords.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel

class User(BaseModel):
    username: str
    email: str
    hashed_password: str

class UserOut(BaseModel):
    username: str
    email: str

app = FastAPI()

@app.get("/users/me", response_model=UserOut)
def read_user():
    return User(username="bob", email="[email protected]", hashed_password="secret")

# response: {"username": "bob", "email": "[email protected]"}  (no password)

Exclude Fields

response_model_include and response_model_exclude let you whitelist or blacklist fields per route without defining new models. Pass a set of field names. These are applied in addition to the response_model.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    secret: str

app = FastAPI()

@app.get("/items/{id}", response_model=Item, response_model_exclude={"secret"})
def read_item(id: int):
    return {"name": "Apple", "price": 0.5, "secret": "hidden"}

@app.get("/items", response_model=list[Item], response_model_include={"name", "price"})
def list_items():
    return [{"name": "Apple", "price": 0.5, "secret": "x"}]

List Response

Use response_model=list[Model] (or List[Model]) to declare that an endpoint returns an array of objects. FastAPI validates and serializes each item in the list.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel

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

app = FastAPI()

@app.get("/items", response_model=list[Item])
def list_items():
    return [
        {"name": "Apple", "price": 0.5},
        {"name": "Banana", "price": 0.3},
    ]

# response: [{"name": "Apple", "price": 0.5}, {"name": "Banana", "price": 0.3}]

Exclude Unset/Defaults

exclude_unset omits fields the client never sent (only those explicitly set appear). exclude_defaults omits fields equal to their defaults; exclude_none omits fields that are None. Useful for sparse responses.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel

class Item(BaseModel):
    name: str
    description: str | None = None
    price: float = 0.0
    tax: float = 0.0

app = FastAPI()

@app.get("/items/{id}", response_model=Item, response_model_exclude_unset=True)
def read_item(id: int):
    return {"name": "Apple", "price": 0.5}
# -> {"name": "Apple", "price": 0.5}   (description, tax omitted)

# also: response_model_exclude_defaults=True
#       response_model_exclude_none=True

Response by Alias

Field(alias=...) lets incoming JSON use a different key name than the Python attribute. response_model_by_alias=True outputs using the alias too. populate_by_name=True also allows the original attribute name on input.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel, ConfigDict, Field

class Item(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    name: str = Field(alias="itemName")
    price: float

app = FastAPI()

@app.post("/items", response_model=Item, response_model_by_alias=True)
def create(item: Item):
    return item

# input:  {"itemName": "Apple", "price": 0.5}
# output: {"itemName": "Apple", "price": 0.5}

Plain Dict Response

Returning a dict/list auto-serializes to JSON. Use response_class to change the default response type (HTMLResponse, PlainTextResponse), or return a Response subclass directly for full control.

fastapi
from fastapi import FastAPI
from fastapi.responses import JSONResponse, PlainTextResponse

app = FastAPI()

@app.get("/dict")
def dict_resp():
    return {"a": 1, "b": 2}      # auto JSON

@app.get("/text", response_class=PlainTextResponse)
def text_resp():
    return "plain text response"

@app.get("/raw")
def raw_resp():
    return JSONResponse(content={"a": 1}, status_code=200)
06

Form & UploadFile

Form Data

Form() declares a parameter parsed from application/x-www-form-urlencoded or multipart form data. You cannot mix Form() with a JSON body (BaseModel) in the same operation — pick one content type.

fastapi
from fastapi import FastAPI, Form

app = FastAPI()

@app.post("/login")
def login(username: str = Form(), password: str = Form()):
    return {"username": username}

# Content-Type: application/x-www-form-urlencoded
# body: username=bob&password=secret

UploadFile Basic

UploadFile is a streaming file representation with .filename, .content_type, .read(), .write(), .close(). Use 'await file.read()' inside async def. For large files prefer file.file (SpooledTemporaryFile) to avoid loading all into memory.

fastapi
from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/upload")
async def upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    return {
        "filename": file.filename,
        "size": len(contents),
        "content_type": file.content_type,
    }

Multiple Files

Declare list[UploadFile] to accept multiple files in one request. Each file is sent with the same field name. The order of files in the list matches the order they were sent.

fastapi
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/uploads")
async def upload_many(files: list[UploadFile]):
    return [
        {"filename": f.filename, "size": len(await f.read())}
        for f in files
    ]

# curl: curl -F "[email protected]" -F "[email protected]" http://localhost:8000/uploads

File and Form Together

Form() and UploadFile (File()) can be combined in the same endpoint because both use multipart/form-data. You still cannot combine them with a JSON BaseModel body.

fastapi
from fastapi import FastAPI, UploadFile, File, Form

app = FastAPI()

@app.post("/items")
async def create_item(
    name: str = Form(),
    description: str = Form(),
    file: UploadFile = File(),
):
    return {
        "name": name,
        "description": description,
        "file": file.filename,
    }

Save Uploaded File

For small files, shutil.copyfileobj streams the SpooledTemporaryFile to disk efficiently. For very large files, read in chunks (e.g. 1MB) to keep memory low. Always sanitize file.filename to avoid path traversal.

fastapi
import shutil
from pathlib import Path
from fastapi import FastAPI, UploadFile

app = FastAPI()

@app.post("/save")
async def save_file(file: UploadFile):
    dest = Path("uploads") / file.filename
    dest.parent.mkdir(exist_ok=True)
    with dest.open("wb") as buf:
        shutil.copyfileobj(file.file, buf)
    return {"saved_to": str(dest)}

# async alternative for large files:
# async with dest.open("wb") as buf:
#     while chunk := await file.read(1024 * 1024):
#         buf.write(chunk)

File Metadata & Size

Validate file.content_type and size before processing. After read(), the cursor is at the end; call 'await file.seek(0)' to rewind if you need to read again. Real type checking should inspect magic bytes, not just the header.

fastapi
from fastapi import FastAPI, UploadFile, File

app = FastAPI()

@app.post("/validate")
async def validate(file: UploadFile = File(...)):
    # allowed types
    allowed = {"image/png", "image/jpeg"}
    if file.content_type not in allowed:
        return {"error": "unsupported type"}

    contents = await file.read()
    max_size = 5 * 1024 * 1024  # 5 MB
    if len(contents) > max_size:
        return {"error": "file too large"}

    await file.seek(0)  # rewind so later reads work
    return {"filename": file.filename, "size": len(contents)}
07

Headers & Cookies

Header Parameters

Header() declares a parameter read from request headers. FastAPI converts underscores in Python names to hyphens (x_token -> X-Token). Set convert_underscores=False to disable this behavior.

fastapi
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/items")
def read_items(user_agent: str | None = Header(default=None)):
    return {"User-Agent": user_agent}

# Header names are case-insensitive; an underscore in the
# parameter (x_token) is converted to a hyphen (X-Token) automatically.
@app.get("/secure")
def secure(x_token: str = Header()):
    return {"x-token": x_token}

Cookie Parameters

Cookie() reads a value from the request's Cookie header. Like query params, cookies without a default are required, and you can apply validation (min_length, max_length, etc.).

fastapi
from fastapi import FastAPI, Cookie

app = FastAPI()

@app.get("/session")
def read_session(session_id: str | None = Cookie(default=None)):
    if session_id:
        return {"session_id": session_id}
    return {"message": "no session"}

# Reads the 'session_id' cookie from the Cookie header.

Set Cookie in Response

Inject a Response and call set_cookie() to write cookies on the outgoing response. Use httponly=True (block JS access), secure=True (HTTPS only), and samesite='lax'/'strict' to mitigate CSRF. The response body is still returned normally.

fastapi
from fastapi import FastAPI, Response

app = FastAPI()

@app.post("/login")
def login(response: Response):
    response.set_cookie(
        key="session_id",
        value="abc123",
        httponly=True,
        max_age=1800,
        secure=True,
        samesite="lax",
    )
    return {"message": "logged in"}

@app.post("/logout")
def logout(response: Response):
    response.delete_cookie("session_id")
    return {"message": "logged out"}

Custom Response Headers

Set headers on an injected Response to add them to whatever JSON response FastAPI builds, or pass headers= when returning a Response subclass directly. Custom headers are commonly used for rate-limit info and process timing.

fastapi
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse

app = FastAPI()

@app.get("/data")
def get_data(response: Response):
    response.headers["X-Process-Time"] = "0.05"
    response.headers["X-Custom-Header"] = "hello"
    return {"data": 1}

@app.get("/direct")
def direct():
    return JSONResponse(
        content={"data": 1},
        headers={"X-Source": "direct"},
    )

Standard Headers

Common headers like Host, Accept, Authorization, If-None-Match are accessible via Header(). For Authorization, prefer the built-in security utilities (HTTPBearer, OAuth2) which handle parsing and OpenAPI documentation for you.

fastapi
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/info")
def info(
    host: str | None = Header(default=None),
    accept: str | None = Header(default=None),
    authorization: str | None = Header(default=None),
    if_none_match: str | None = Header(default=None),
):
    return {"host": host, "accept": accept}

Duplicate Header Values

Declaring a list type for a Header parameter collects all repeated header values into a list. This is useful for headers like Set-Cookie (response) or any custom header that may appear multiple times.

fastapi
from fastapi import FastAPI, Header

app = FastAPI()

@app.get("/multi")
def multi(x_token: list[str] | None = Header(default=None)):
    return {"x_token": x_token}

# Request with:
#   X-Token: a
#   X-Token: b
# -> {"x_token": ["a", "b"]}
08

Dependency Injection

Basic Depends

Depends() injects the return value of a function into your path operation. Shared logic (pagination, parsing, DB sessions) lives in one place and is reused across endpoints. The dependency's parameters are also documented in OpenAPI.

fastapi
from fastapi import FastAPI, Depends

app = FastAPI()

def common_params(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items")
def list_items(commons: dict = Depends(common_params)):
    return commons

@app.get("/users")
def list_users(commons: dict = Depends(common_params)):
    return commons

Class Dependencies

A class with an __init__ can be a dependency; FastAPI calls it like a function. Use Depends(CommonQueryParams) or the shorthand Depends() when the parameter is annotated with the class. The instance is injected.

fastapi
from fastapi import FastAPI, Depends

class CommonQueryParams:
    def __init__(self, q: str | None = None, skip: int = 0, limit: int = 10):
        self.q = q
        self.skip = skip
        self.limit = limit

app = FastAPI()

@app.get("/items")
def list_items(commons: CommonQueryParams = Depends(CommonQueryParams)):
    return {"q": commons.q, "skip": commons.skip, "limit": commons.limit}

# shorthand: commons = Depends() infers the type

Nested Dependencies

Dependencies can themselves depend on other dependencies, forming a graph. FastAPI resolves the whole tree, caches each result per request, and propagates exceptions. This is ideal for layered auth (token -> user -> permissions).

fastapi
from fastapi import FastAPI, Depends, Header, HTTPException

def get_token(authorization: str = Header()):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="invalid header")
    return authorization.removeprefix("Bearer ")

def get_current_user(token: str = Depends(get_token)):
    if token != "admin-token":
        raise HTTPException(status_code=403, detail="forbidden")
    return {"username": "admin"}

app = FastAPI()

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

Yield Dependencies (Cleanup)

A dependency generator (yield) is perfect for resources needing cleanup: DB sessions, file handles, transactions. Code after yield runs as a finalizer, even if the route raises an exception. Only one yield per dependency.

fastapi
from fastapi import FastAPI, Depends

def get_db():
    db = open_db_session()       # acquire resource
    try:
        yield db                 # injected into the route
    finally:
        db.close()               # always runs after the request

app = FastAPI()

@app.get("/items")
def list_items(db = Depends(get_db)):
    return db.query("SELECT * FROM items")

Global Dependencies

Pass dependencies=[...] to FastAPI() (or an APIRouter) to enforce them on every route without injecting their return value. Useful for global auth, rate limiting, or request logging.

fastapi
from fastapi import FastAPI, Depends, Header, HTTPException

def verify_token(x_token: str = Header()):
    if x_token != "supersecret":
        raise HTTPException(status_code=400, detail="X-Token missing")

def verify_key(x_key: str = Header()):
    if x_key != "realkey":
        raise HTTPException(status_code=400, detail="X-Key missing")

# applied to every route in the app
app = FastAPI(dependencies=[Depends(verify_token), Depends(verify_key)])

@app.get("/items")
def list_items():
    return [{"item": "a"}]   # token & key already checked

Per-Route Dependencies

Use dependencies=[Depends(...)] on a decorator (or APIRouter) to require a dependency for specific routes without using its return value. This keeps route signatures clean while still enforcing checks.

fastapi
from fastapi import FastAPI, Depends, Header, HTTPException

def verify_token(x_token: str = Header()):
    if x_token != "secret":
        raise HTTPException(status_code=401, detail="bad token")

app = FastAPI()

@app.get("/public")
def public():
    return {"open": True}

@app.get("/private", dependencies=[Depends(verify_token)])
def private():
    return {"secret": "data"}
09

Security (HTTPBasic / OAuth2 / JWT)

HTTP Basic Auth

HTTPBasic parses the Authorization: Basic header into username/password. Use secrets.compare_digest (not ==) to prevent timing attacks. Always return 401 with WWW-Authenticate: Basic so browsers prompt for credentials.

fastapi
import secrets
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials

security = HTTPBasic()
app = FastAPI()

def get_current_user(credentials: HTTPBasicCredentials = Depends(security)):
    correct_user = secrets.compare_digest(credentials.username, "admin")
    correct_pass = secrets.compare_digest(credentials.password, "secret")
    if not (correct_user and correct_pass):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid credentials",
            headers={"WWW-Authenticate": "Basic"},
        )
    return credentials.username

@app.get("/profile")
def profile(user: str = Depends(get_current_user)):
    return {"user": user}

OAuth2 Password Bearer

OAuth2PasswordBearer declares that routes expect a Bearer token; the dependency extracts it from the Authorization header. tokenUrl tells the docs UI where to send credentials. The token itself is just a string until you validate it.

fastapi
from fastapi import FastAPI, Depends
from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
app = FastAPI()

@app.get("/me")
def me(token: str = Depends(oauth2_scheme)):
    return {"token": token}

# client POSTs username+password to /token, receives an access token,
# then sends it as: Authorization: Bearer <token>

Create JWT Token

Use python-jose (or PyJWT) to sign tokens. The 'sub' (subject) claim conventionally holds the username or user id, and 'exp' sets expiry. Keep SECRET_KEY long, random, and loaded from environment — never commit it.

fastapi
from datetime import datetime, timedelta, timezone
from jose import jwt

SECRET_KEY = "keep-this-secret"
ALGORITHM = "HS256"

def create_access_token(data: dict, expires_minutes: int = 30) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=expires_minutes)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

# token = create_access_token({"sub": "alice"})
# -> eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...

Decode JWT Token

jwt.decode verifies the signature and the 'exp' claim automatically, raising JWTError on failure. Map the dependency to a 401 with WWW-Authenticate: Bearer so clients know to provide a valid token.

fastapi
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
SECRET_KEY = "keep-this-secret"
ALGORITHM = "HS256"

def get_current_user(token: str = Depends(oauth2_scheme)) -> str:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str | None = payload.get("sub")
        if username is None:
            raise credentials_exception
        return username
    except JWTError:
        raise credentials_exception

credentials_exception = HTTPException(
    status_code=status.HTTP_401_UNAUTHORIZED,
    detail="Could not validate credentials",
    headers={"WWW-Authenticate": "Bearer"},
)

Password Hashing

Never store plaintext passwords. passlib with bcrypt (or argon2) handles salting and hashing. verify() compares a plaintext input against a stored hash. pip install passlib[bcrypt] to get the bcrypt backend.

fastapi
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def hash_password(plain: str) -> str:
    return pwd_context.hash(plain)

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

# usage when issuing tokens:
def authenticate_user(username: str, password: str):
    user = fake_db.get(username)
    if not user or not verify_password(password, user.hashed_password):
        return False
    return user

OAuth2 Login Flow

OAuth2PasswordRequestForm reads username/password from a form-encoded POST (the standard OAuth2 password flow). Return access_token and token_type: 'bearer'. This endpoint is what tokenUrl in OAuth2PasswordBearer points to.

fastapi
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm

app = FastAPI()

@app.post("/token")
def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate_user(form.username, form.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect username or password",
        )
    token = create_access_token({"sub": user.username})
    return {"access_token": token, "token_type": "bearer"}

# OAuth2PasswordRequestForm parses username+password from form data
# and powers the 'Authorize' button in the Swagger UI.
10

Middleware

Add Middleware

@app.middleware('http') registers a function that wraps every request. It must await call_next(request) to get the response, then can inspect or modify it before returning. Middleware runs for all routes.

fastapi
import time
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response

Custom Header Middleware

Middleware is the right place to attach headers that should be on every response (security headers, server banner, tracing IDs). Mutate the response object after awaiting call_next.

fastapi
from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def add_banner(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Powered-By"] = "FastAPI"
    response.headers["X-Frame-Options"] = "DENY"
    return response

# every response now carries these headers

BaseHTTPMiddleware

For reusable, configurable middleware, subclass BaseHTTPMiddleware and register it with app.add_middleware(MyMiddleware, **options). This is also how the built-in CORS, GZip, and TrustedHost middlewares are added.

fastapi
from starlette.middleware.base import BaseHTTPMiddleware
from fastapi import FastAPI, Request

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start = time.perf_counter()
        response = await call_next(request)
        response.headers["X-Duration"] = str(time.perf_counter() - start)
        return response

app = FastAPI()
app.add_middleware(TimingMiddleware)

# class-based middleware is reusable and configurable

HTTPS Redirect

HTTPSRedirectMiddleware returns a 308 redirect to the HTTPS version of any HTTP request. Use it behind a load balancer that terminates TLS, to guarantee clients upgrade to a secure connection.

fastapi
from fastapi import FastAPI
from fastapi.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()
app.add_middleware(HTTPSRedirectMiddleware)

# all HTTP requests are redirected to HTTPS (308)

Trusted Host

TrustedHostMiddleware rejects requests whose Host header is not in the allow-list, preventing Host-header injection attacks. Wildcards like '*.example.com' are supported. Always include 'localhost' for local development.

fastapi
from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware

app = FastAPI()
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com", "localhost"],
)

# requests with a disallowed Host header get a 400 response

Middleware Order

Starlette executes middleware in LIFO order for incoming requests: the last added runs first. Add CORS early (so preflight works) and GZip after. Remember this when debugging why a header set by one middleware isn't seen by another.

fastapi
from fastapi import FastAPI
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Middleware runs in REVERSE order of registration for requests,
# and in registration order for responses.
app.add_middleware(GZipMiddleware, minimum_size=1000)
app.add_middleware(CORSMiddleware, allow_origins=["*"])

# request flow:  CORS -> GZip -> route
# response flow: route -> GZip -> CORS
11

CORS

CORSMiddleware Setup

CORSMiddleware handles the CORS preflight (OPTIONS) and adds Access-Control-* headers. allow_origins should list the exact origins permitted; use a list of specific URLs in production rather than ['*'].

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

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

@app.get("/api/data")
def data():
    return {"data": 1}

Allow All Origins (Dev Only)

allow_origins=['*'] permits any origin but cannot be combined with allow_credentials=True (browsers reject it). Use this only for fully public APIs in development; in production enumerate the real origins.

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["*"],
    allow_headers=["*"],
    # allow_credentials=True CANNOT be combined with allow_origins=["*"]
)

Allow Credentials

allow_credentials=True lets the browser send cookies and Authorization headers cross-origin. It requires specific origins (never '*'). The browser sets Access-Control-Allow-Credentials: true on responses.

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_credentials=True,    # send cookies / Authorization
    allow_methods=["GET", "POST"],
    allow_headers=["Authorization", "Content-Type"],
)

# needed when the frontend sends cookies or credentials with requests

Allowed Methods & Headers

allow_methods/allow_headers restrict what the browser may send. expose_headers lists response headers the frontend JavaScript is allowed to read (browsers otherwise hide custom headers). Preflight responses are cached by the browser.

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
    allow_headers=["Authorization", "Content-Type", "X-Request-ID"],
    expose_headers=["X-Total-Count", "X-Process-Time"],
)

# expose_headers lets the frontend JS read these response headers

Preflight Requests

For 'non-simple' requests (e.g. custom headers, PUT/DELETE, JSON content type), the browser sends an OPTIONS preflight. CORSMiddleware intercepts OPTIONS and responds with the allowed methods/headers. max_age controls how long the browser caches that answer.

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],
    allow_methods=["*"],
    allow_headers=["*"],
    max_age=600,   # browser caches preflight result for 600s
)

# Browser sends OPTIONS preflight for non-simple requests;
# CORSMiddleware answers automatically with the right headers.

Dynamic Origin Validation

For dynamic allow-lists use allow_origin_regex (matches against the request Origin) instead of a static list. This is handy when valid origins follow a pattern (e.g. any subdomain of example.com).

fastapi
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

VALID_DOMAINS = {"example.com", "staging.example.com"}

def origin_check(origin: str) -> str | None:
    if origin and origin.replace("https://", "").replace("http://", "") in VALID_DOMAINS:
        return origin
    return None

app.add_middleware(
    CORSMiddleware,
    allow_origin_regex=r"https://.*\.example\.com",
    allow_methods=["*"],
    allow_headers=["*"],
)
12

Exception Handling

Raise HTTPException

raise HTTPException anywhere in a route or dependency to short-circuit and return an error response. detail can be a string, dict, or list — it becomes the JSON 'detail' field. headers optionally adds response headers.

fastapi
from fastapi import FastAPI, HTTPException, status

app = FastAPI()

items = {1: "Apple", 2: "Banana"}

@app.get("/items/{item_id}")
def read_item(item_id: int):
    if item_id not in items:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="Item not found",
            headers={"X-Error": "missing"},
        )
    return {"item": items[item_id]}

Custom Exception

Register a handler with @app.exception_handler(MyException) to convert any custom exception into an HTTP response. The handler receives the request and the exception. This centralizes error formatting for your app's domain errors.

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

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops, {exc.name} did it again"},
    )

@app.get("/unicorns/{name}")
def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"name": name}

Override Validation Error

RequestValidationError is raised when input fails Pydantic validation. Override its handler with @app.exception_handler(RequestValidationError) to customize the 422 response shape, e.g. to match a company-wide error envelope.

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

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "success": False,
            "errors": exc.errors(),
            "body": exc.body,
        },
    )

Override HTTPException Handler

Override the default HTTPException handler to enforce a consistent error JSON shape across all raised HTTPExceptions. Import the class as fastapi.exceptions.HTTPException (the same one used by raise) when re-binding its handler.

fastapi
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import HTTPException as FastAPIHTTPException

app = FastAPI()

@app.exception_handler(FastAPIHTTPException)
async def custom_http_handler(request, exc: FastAPIHTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "success": False,
            "status": exc.status_code,
            "message": exc.detail,
        },
        headers=exc.headers,
    )

Common Status Codes

The status module gives named constants for all HTTP status codes, avoiding magic numbers. 200 (default GET), 201 (created), 204 (no content), 422 (validation) are the most common in FastAPI apps.

fastapi
from fastapi import FastAPI, status

app = FastAPI()

@app.get("/ok", status_code=status.HTTP_200_OK)
def ok(): return {"ok": True}

@app.post("/create", status_code=status.HTTP_201_CREATED)
def create(): return {"id": 1}

@app.get("/no-content", status_code=status.HTTP_204_NO_CONTENT)
def no_content(): return None

@app.get("/not-found")
def missing():
    # raise HTTPException(status.HTTP_404_NOT_FOUND, "nope")
    ...

# Other common: 400 BAD_REQUEST, 401 UNAUTHORIZED,
# 403 FORBIDDEN, 409 CONFLICT, 422 UNPROCESSABLE_ENTITY, 500 INTERNAL_SERVER_ERROR

Generic Exception Handler

Registering a handler for the base Exception catches anything not already handled, preventing stack traces from leaking to clients. Always log the full traceback server-side; return a generic 500 to the client.

fastapi
import logging
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

logger = logging.getLogger("uvicorn.error")
app = FastAPI()

@app.exception_handler(Exception)
async def unhandled(request: Request, exc: Exception):
    logger.exception("Unhandled error on %s", request.url.path)
    return JSONResponse(
        status_code=500,
        content={"detail": "Internal Server Error"},
    )

@app.get("/boom")
def boom():
    raise RuntimeError("something exploded")
13

Background Tasks

Basic BackgroundTasks

BackgroundTasks runs functions after the response is sent — perfect for fire-and-forget work like logging, emails, or cache invalidation. add_task schedules a function (and args) without blocking the response.

fastapi
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(message + "\n")

@app.post("/send")
def send_notification(background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, "notification sent")
    return {"message": "notification queued"}

Multiple Tasks

Add as many tasks as you want; they run sequentially in the order added, after the response is returned. Because they run in the same process, keep them quick and non-blocking, or move to a real queue (Celery, RQ, Dramatiq).

fastapi
from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, subject: str): ...
def update_cache(key: str): ...
def notify_slack(msg: str): ...

@app.post("/signup")
def signup(background_tasks: BackgroundTasks):
    background_tasks.add_task(send_email, "[email protected]", "Welcome")
    background_tasks.add_task(update_cache, "user_count")
    background_tasks.add_task(notify_slack, "new signup")
    return {"status": "ok"}

Task with Dependencies

BackgroundTasks is injected just like any dependency. The task function can receive values (like a DB session) captured at request time. Note that yielded/closed resources from dependencies close after tasks finish, since tasks run within the request lifecycle's scope.

fastapi
from fastapi import FastAPI, BackgroundTasks, Depends

app = FastAPI()

def get_db_session():
    db = "session"
    yield db

def write_log(db, message: str):
    print(db, message)

@app.get("/items")
def list_items(background_tasks: BackgroundTasks, db = Depends(get_db_session)):
    background_tasks.add_task(write_log, db, "list called")
    return {"items": []}

Async Task Function

Background task functions can be either sync (run in a threadpool) or async (awaited on the event loop). Use async tasks for I/O-bound work that itself uses async libraries, and sync tasks for CPU/blocking work.

fastapi
from fastapi import FastAPI, BackgroundTasks
import httpx

app = FastAPI()

async def fetch_url(url: str):
    async with httpx.AsyncClient() as client:
        r = await client.get(url)
        return r.status_code

@app.post("/webhook")
def webhook(url: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(fetch_url, url)
    return {"queued": url}

Background vs Celery

Use BackgroundTasks for lightweight, non-critical work that is fine to lose on restart. For jobs that must survive crashes, retry, or scale across machines, use a real task queue (Celery, RQ, Dramatiq, Arq) with a broker like Redis or RabbitMQ.

fastapi
# BackgroundTasks: in-process, no broker, lost if process crashes
@app.post("/quick")
def quick(tasks: BackgroundTasks):
    tasks.add_task(small_job)
    return {"ok": True}

# Celery/RQ/Dramatiq: separate workers, persistent, retryable
# from celery import Celery
# @celery.task
# def big_job(payload): ...
#
# @app.post("/heavy")
# def heavy(payload: dict):
#     big_job.delay(payload)
#     return {"queued": True}

Email Sending Example

Email sending is the classic BackgroundTasks use case: it's slow and the user shouldn't wait. The response returns immediately while the email is dispatched after. For high volume, prefer a queue and an email-service SDK with retries.

fastapi
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, EmailStr

class Message(BaseModel):
    to: EmailStr
    subject: str
    body: str

app = FastAPI()

def send_email(to: str, subject: str, body: str):
    # smtplib or your provider's SDK
    import smtplib
    with smtplib.SMTP("localhost") as s:
        s.sendmail("[email protected]", to, f"Subject: {subject}\n\n{body}")

@app.post("/contact")
def contact(msg: Message, tasks: BackgroundTasks):
    tasks.add_task(send_email, msg.to, msg.subject, msg.body)
    return {"status": "queued", "to": msg.to}
14

WebSocket

WebSocket Basic

Use @app.websocket() to define a WebSocket endpoint. You must accept() the connection before sending/receiving. The function typically loops reading messages; receive_text/receive_bytes/receive_json all await data from the client.

fastapi
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    while True:
        data = await ws.receive_text()
        await ws.send_text(f"echo: {data}")

# client (browser):
# const ws = new WebSocket("ws://localhost:8000/ws");
# ws.onmessage = (e) => console.log(e.data);

Send JSON

send_json() and receive_json() handle serialization for you, exchanging JSON objects over the socket. For binary data use send_bytes/receive_bytes. Each send/receive is a separate frame on the WebSocket.

fastapi
import json
from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    await ws.accept()
    while True:
        msg = await ws.receive_json()
        await ws.send_json({"reply": msg.get("text", "").upper()})

# send_text / send_bytes / send_json helpers are available

WebSocket Close

WebSocketDisconnect is raised when the client closes the connection — catch it to clean up. Call ws.close(code=1000) for a normal close; common codes are 1000 (normal), 1008 (policy violation), 1011 (server error).

fastapi
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

app = FastAPI()

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            if data == "quit":
                await ws.close(code=1000)
                break
            await ws.send_text(data)
    except WebSocketDisconnect:
        print("client disconnected")

Connection Manager

A ConnectionManager tracks active sockets so you can broadcast to all connected clients (e.g. a chat room). Always remove the socket on disconnect to avoid sending to closed connections and to keep memory bounded.

fastapi
from fastapi import FastAPI, WebSocket, WebSocketDisconnect

class ConnectionManager:
    def __init__(self):
        self.active: list[WebSocket] = []

    async def connect(self, ws: WebSocket):
        await ws.accept()
        self.active.append(ws)

    def disconnect(self, ws: WebSocket):
        self.active.remove(ws)

    async def broadcast(self, message: str):
        for ws in self.active:
            await ws.send_text(message)

manager = ConnectionManager()
app = FastAPI()

@app.websocket("/chat")
async def chat(ws: WebSocket):
    await manager.connect(ws)
    try:
        while True:
            msg = await ws.receive_text()
            await manager.broadcast(msg)
    except WebSocketDisconnect:
        manager.disconnect(ws)

WebSocket Auth via Query Param

Browsers cannot set custom headers on WebSocket connections, so auth tokens are commonly passed as query parameters (or cookies). Validate before calling accept(), and close with code 1008 (policy violation) if unauthorized.

fastapi
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException

app = FastAPI()

async def authenticate(token: str) -> str:
    if token != "secret":
        raise HTTPException(status_code=401)
    return "alice"

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket):
    token = ws.query_params.get("token")
    if not token:
        await ws.close(code=1008)
        return
    user = await authenticate(token)
    await ws.accept()
    try:
        while True:
            data = await ws.receive_text()
            await ws.send_text(f"{user}: {data}")
    except WebSocketDisconnect:
        pass

WebSocket with Dependencies

Dependencies can be used in WebSocket endpoints just like in HTTP routes, but only dependencies that read from the initial handshake (headers, query params, cookies) make sense — there is no body. Inject them via Depends() as usual.

fastapi
from fastapi import FastAPI, WebSocket, Depends, Header

app = FastAPI()

def token_validator(token: str = Header(default="")):
    # for WS, headers come from the initial HTTP handshake
    return token == "valid"

@app.websocket("/ws")
async def ws_endpoint(ws: WebSocket, valid: bool = Depends(token_validator)):
    if not valid:
        await ws.close(code=1008)
        return
    await ws.accept()
    while True:
        msg = await ws.receive_text()
        await ws.send_text(msg)
15

Database (SQLAlchemy)

SQLAlchemy Setup

create_engine sets up the DB connection pool; SessionLocal is a factory for sessions; Base is the declarative base for models. For SQLite with FastAPI, set check_same_thread=False because requests may run in different threads.

fastapi
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./app.db"
# postgresql: "postgresql://user:pass@localhost:5432/dbname"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    connect_args={"check_same_thread": False},  # sqlite only
)
SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
Base = declarative_base()

Session Dependency

A yield dependency is the canonical way to provide a DB session: each request gets a fresh session that is closed in the finally block. This guarantees cleanup even if the route raises an exception.

fastapi
from fastapi import Depends
from sqlalchemy.orm import Session

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get("/items")
def list_items(db: Session = Depends(get_db)):
    return db.query(Item).all()

# each request gets its own session, closed automatically afterward

Model Definition

Define tables as SQLAlchemy ORM models subclassing Base. Use Alembic for schema migrations in production rather than create_all(). Columns can be indexed, nullable, and have defaults applied at the database or Python level.

fastapi
from sqlalchemy import Column, Integer, String, Boolean
from sqlalchemy.orm import declarative_base

Base = declarative_base()

class Item(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True, nullable=False)
    price = Column(Integer, default=0)
    in_stock = Column(Boolean, default=True)

# create tables (for dev; use Alembic migrations in prod)
Base.metadata.create_all(bind=engine)

Create Record

add() stages a new row, commit() writes it to the DB, and refresh() reloads the instance with server-generated values (like the auto-increment id). Wrap multiple writes in a transaction and commit once at the end.

fastapi
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

@app.post("/items", status_code=201)
def create_item(payload: ItemCreate, db: Session = Depends(get_db)):
    item = Item(name=payload.name, price=payload.price)
    db.add(item)
    db.commit()
    db.refresh(item)        # load generated id
    return item

Query Records

db.query(Model) starts a query; .filter() applies WHERE conditions, .offset()/.limit() paginate, .first() returns one row or None, .all() returns a list. Use .filter_by(name=x) for simple equality conditions.

fastapi
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

@app.get("/items")
def list_items(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
    return db.query(Item).offset(skip).limit(limit).all()

@app.get("/items/{item_id}")
def get_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="not found")
    return item

Update & Delete

For partial updates, iterate model_dump(exclude_unset=True) so only provided fields change. delete() stages removal; commit() finalizes it. Return 204 No Content for DELETE on success.

fastapi
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session

app = FastAPI()

@app.put("/items/{item_id}")
def update_item(item_id: int, payload: ItemUpdate, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if not item:
        raise HTTPException(status_code=404, detail="not found")
    for k, v in payload.model_dump(exclude_unset=True).items():
        setattr(item, k, v)
    db.commit()
    db.refresh(item)
    return item

@app.delete("/items/{item_id}", status_code=204)
def delete_item(item_id: int, db: Session = Depends(get_db)):
    item = db.query(Item).filter(Item.id == item_id).first()
    if item:
        db.delete(item)
        db.commit()
16

Database (Tortoise ORM)

Tortoise Setup

Tortoise ORM is async-native, pairing well with FastAPI. Init it at startup with db_url and the module(s) containing your models. generate_schemas() is fine for dev; use Aerich for migrations in production. Use the lifespan handler instead of on_event in new code.

fastapi
from fastapi import FastAPI
from tortoise import Tortoise

app = FastAPI()

async def init():
    await Tortoise.init(
        db_url="sqlite://db.sqlite3",
        modules={"models": ["app.models"]},
    )
    await Tortoise.generate_schemas()

@app.on_event("startup")
async def startup():
    await init()

@app.on_event("shutdown")
async def shutdown():
    await Tortoise.close_connections()

Model Definition

Tortoise models subclass Model and declare fields via tortoise.fields. Field types map to DB column types. Set the table name in the inner Meta class. Relationships use ForeignKeyField, ManyToManyField, etc.

fastapi
from tortoise import fields
from tortoise.models import Model

class Item(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=100)
    price = fields.FloatField(default=0.0)
    created_at = fields.DatetimeField(auto_now_add=True)

    class Meta:
        table = "items"

    def __str__(self):
        return self.name

Create Record

Item.create() is a coroutine that inserts and returns the new row. To return a Pydantic response, build a BaseModel with from_tortoise_orm() (via tortoise.contrib.pydantic) so Tortoise serializes async-loaded relations correctly.

fastapi
from fastapi import FastAPI
from pydantic import BaseModel
from .models import Item

app = FastAPI()

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

@app.post("/items", response_model=ItemOut, status_code=201)
async def create_item(payload: ItemIn):
    item = await Item.create(name=payload.name, price=payload.price)
    return await ItemOut.from_tortoise_orm(item)

Async Query

Tortoise queries are awaited: .all() returns a queryset, .filter() adds conditions, .get_or_none() fetches one row or None, .first() returns the first match. Avoid .get() unless you expect exactly one row (it raises if not found).

fastapi
from fastapi import FastAPI, HTTPException
from .models import Item

app = FastAPI()

@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    return await Item.all().offset(skip).limit(limit)

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

Update & Delete

Modify model attributes then await save() to update a row, or await Item.filter(id=...).update(name=x) for bulk updates. await item.delete() removes the row. Each method is a coroutine and must be awaited.

fastapi
from fastapi import FastAPI, HTTPException
from .models import Item

app = FastAPI()

@app.put("/items/{item_id}")
async def update_item(item_id: int, name: str):
    item = await Item.get_or_none(id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="not found")
    item.name = name
    await item.save()
    return item

@app.delete("/items/{item_id}", status_code=204)
async def delete_item(item_id: int):
    item = await Item.get_or_none(id=item_id)
    if item:
        await item.delete()

Tortoise Pydantic Schemas

tortoise.contrib.pydantic generates Pydantic models directly from Tortoise models, so you don't maintain two schemas. Use pydantic_model_creator for a single object and pydantic_queryset_creator for a list response.

fastapi
from tortoise.contrib.pydantic import pydantic_model_creator, pydantic_queryset_creator
from .models import Item

ItemOut = pydantic_model_creator(Item, name="ItemOut")
ItemList = pydantic_queryset_creator(Item)

@app.get("/items", response_model=ItemList)
async def list_items():
    return await Item.all()

# auto-generates a Pydantic model from the Tortoise model,
# including types and docstrings.
17

Testing (TestClient)

TestClient Basic

TestClient wraps your app and lets you make requests in-process (no real socket). It's based on httpx and works with pytest out of the box. Import the same app instance you run in production.

fastapi
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_root():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"Hello": "World"}

# TestClient is based on httpx; no network is involved —
# it calls the ASGI app directly in-process.

Test GET & POST

Pass json= for JSON bodies, params= for query params, headers= for headers, and files= for multipart uploads. Validation failures return 422 — assert on status_code and the structure of the error 'detail'.

fastapi
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_create_item():
    response = client.post(
        "/items",
        json={"name": "Apple", "price": 0.5},
    )
    assert response.status_code == 201
    assert response.json()["name"] == "Apple"

def test_validation_error():
    response = client.post("/items", json={"name": "Apple"})  # missing price
    assert response.status_code == 422

Pytest Fixture

A pytest fixture returning a TestClient keeps tests DRY and lets each test get a fresh client. For DB-backed tests, add a fixture that creates/drops tables or uses a transaction rollback per test to keep them isolated.

fastapi
import pytest
from fastapi.testclient import TestClient
from main import app

@pytest.fixture
def client():
    return TestClient(app)

def test_read(client):
    r = client.get("/items")
    assert r.status_code == 200

def test_create(client):
    r = client.post("/items", json={"name": "X"})
    assert r.status_code == 201

# run: pytest -v

Override Dependencies

app.dependency_overrides swaps a dependency (like the real DB session or auth) for a mock/fake during tests, without changing the routes. This is the canonical way to test endpoints in isolation from external services. Clear it after tests.

fastapi
from fastapi.testclient import TestClient
from main import app, get_db

def get_test_db():
    # return a test session or mock
    yield FakeSession()

app.dependency_overrides[get_db] = get_test_db

client = TestClient(app)

def test_items():
    r = client.get("/items")
    assert r.status_code == 200

# restore afterwards:
app.dependency_overrides.clear()

Test Authentication

OAuth2PasswordRequestForm expects form-encoded data (data=, not json=). Use the returned token in the Authorization: Bearer header for subsequent protected requests. To bypass auth entirely, override the get_current_user dependency instead.

fastapi
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def get_token():
    r = client.post(
        "/token",
        data={"username": "admin", "password": "secret"},
    )
    return r.json()["access_token"]

def test_protected():
    token = get_token()
    r = client.get(
        "/me",
        headers={"Authorization": f"Bearer {token}"},
    )
    assert r.status_code == 200
    assert r.json()["username"] == "admin"

Test WebSocket

TestClient.websocket_connect() opens a WebSocket within a context manager. Use send_text/receive_text (and the json/bytes variants) to exchange messages. Receiving on a closed connection raises WebSocketDisconnect.

fastapi
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_websocket():
    with client.websocket_connect("/ws") as ws:
        ws.send_text("hello")
        data = ws.receive_text()
        assert data == "echo: hello"

def test_websocket_close():
    with client.websocket_connect("/ws") as ws:
        ws.send_text("quit")
        # the server closes the connection
18

OpenAPI Docs

Swagger UI

FastAPI auto-generates an interactive Swagger UI at /docs from your routes, types, and docstrings. Set title/description/version on the app, and add a docstring or summary/description on each route to enrich the UI.

fastapi
from fastapi import FastAPI

app = FastAPI(
    title="My API",
    description="Sample API with auto docs",
    version="1.0.0",
)

@app.get("/items", summary="List items", description="Returns all items")
def list_items():
    """Get every item in the catalog."""
    return [{"id": 1}]

# interactive docs at: http://localhost:8000/docs

ReDoc

ReDoc at /redoc is a read-only, documentation-focused alternative to Swagger. The raw OpenAPI schema lives at /openapi.json and can be fed to other tools. Disable docs with docs_url=None, redoc_url=None, openapi_url=None (e.g. in production).

fastapi
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def root():
    return {"hello": "world"}

# ReDoc (alternative docs UI) at: http://localhost:8000/redoc
# raw OpenAPI JSON at: http://localhost:8000/openapi.json

# disable docs entirely:
# app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)

Tags & Grouping

Tags group routes in the docs UI. Pre-declare tag metadata (name, description, externalDocs) in openapi_tags so the group appears even before any route uses it. Set deprecated=True to mark an endpoint as deprecated in the docs.

fastapi
from fastapi import FastAPI

app = FastAPI(openapi_tags=[
    {"name": "users", "description": "User management"},
    {"name": "items", "description": "Item operations"},
])

@app.get("/users", tags=["users"])
def list_users(): ...

@app.post("/items", tags=["items"])
def create_item(): ...

@app.get("/items/{id}", tags=["items"], deprecated=True)
def get_item(id: int): ...

Route Metadata

Decorators accept metadata that flows into OpenAPI: summary (short title), description (long text), response_description, tags, deprecated, and status_code. A docstring is used as the description if you don't pass one explicitly.

fastapi
from fastapi import FastAPI, status
from pydantic import BaseModel

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

app = FastAPI()

@app.post(
    "/items",
    response_model=Item,
    status_code=status.HTTP_201_CREATED,
    summary="Create an item",
    description="Creates a new item and returns it.",
    tags=["items"],
    response_description="The created item",
)
def create_item(item: Item):
    """Create endpoint docstring — also shown in docs."""
    return item

Field Examples

Field(examples=[...]) and Body(examples=[...]) add example payloads to the Swagger UI's 'Try it out' and example dropdown. Body examples can include multiple cases with summaries to show valid and invalid inputs.

fastapi
from pydantic import BaseModel, Field
from fastapi import FastAPI, Body

class Item(BaseModel):
    name: str = Field(examples=["Apple"])
    price: float = Field(examples=[0.5], gt=0)

app = FastAPI()

@app.post("/items")
def create_item(item: Item):
    return item

@app.post("/manual")
def manual(
    item: Item = Body(
        examples=[
            {"summary": "valid", "value": {"name": "Apple", "price": 0.5}},
            {"summary": "invalid", "value": {"name": "Apple", "price": -1}},
        ],
    ),
):
    return item

Custom OpenAPI Schema

Override app.openapi to customize the generated schema — add a logo, extra metadata, or remove/transform operations. Cache the result in app.openapi_schema so it's only generated once. The schema is what /openapi.json serves.

fastapi
from fastapi import FastAPI
from fastapi.openapi.utils import get_openapi

app = FastAPI()

def custom_openapi():
    if app.openapi_schema:
        return app.openapi_schema
    schema = get_openapi(
        title="Custom API",
        version="2.0.0",
        routes=app.routes,
    )
    schema["info"]["x-logo"] = {"url": "https://example.com/logo.png"}
    app.openapi_schema = schema
    return schema

app.openapi = custom_openapi
19

Deployment (Uvicorn / Gunicorn)

Uvicorn Production

For production, run Uvicorn behind a reverse proxy (Nginx, Caddy, a cloud load balancer). Use --proxy-headers so the app sees the client's real IP/host from X-Forwarded-For/X-Forwarded-Proto. Avoid --reload in production.

fastapi
# Run directly with uvicorn (single worker)
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 1

# Recommended: behind a reverse proxy (nginx) on port 8000
# Use --proxy-headers so X-Forwarded-* is trusted
uvicorn main:app --host 0.0.0.0 --port 8000 --proxy-headers --forwarded-allow-ips='*'

Gunicorn with Uvicorn Workers

Gunicorn manages multiple Uvicorn worker processes, giving you graceful reloads, restarts on crash, and pre-fork scaling. Use the UvicornWorker class. Set workers to (2 * CPU cores + 1) as a starting point, then benchmark.

fastapi
# install the uvicorn worker class
pip install "uvicorn[standard]" gunicorn

# run 4 worker processes managed by gunicorn
gunicorn main:app \
    -w 4 \
    -k uvicorn.workers.UvicornWorker \
    -b 0.0.0.0:8000 \
    --timeout 120

# -w workers (commonly 2*CPU+1), -k worker class

Docker Deployment

A slim Python base image keeps the image small. Install dependencies before copying code to leverage Docker's layer cache. Run with gunicorn+uvicorn workers inside the container, and expose port 8000 to the host or orchestrator.

fastapi
# Dockerfile
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", \
     "-b", "0.0.0.0:8000"]

# build & run:
# docker build -t myapi .
# docker run -p 8000:8000 myapi

Environment Configuration

pydantic-settings (formerly pydantic.BaseSettings) loads config from environment variables and .env files with type validation. This lets the same image run in different environments (dev/staging/prod) just by changing env vars.

fastapi
import os
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str = "sqlite:///./app.db"
    secret_key: str = "change-me"
    debug: bool = False
    allowed_origins: list[str] = ["http://localhost:3000"]

    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

settings = Settings()

# .env file:
# DATABASE_URL=postgresql://user:pass@db:5432/app
# SECRET_KEY=long-random-string
# DEBUG=false

Nginx Reverse Proxy

Nginx terminates HTTP(S) and forwards to Uvicorn on localhost:8000. Set the X-Forwarded-* headers (and use --proxy-headers on Uvicorn) so the app sees the real client IP and scheme. The Upgrade/Connection lines enable WebSocket support.

fastapi
# /etc/nginx/conf.d/api.conf
server {
    listen 80;
    server_name api.example.com;

    location / {
        proxy_pass http://127.0.0.1:8000;
        proxy_set_header Host \$host;
        proxy_set_header X-Real-IP \$remote_addr;
        proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto \$scheme;

        proxy_http_version 1.1;
        proxy_set_header Upgrade \$http_upgrade;
        proxy_set_header Connection "upgrade";
    }
}

Scaling & Concurrency

Scale two ways: more Uvicorn worker processes per machine (CPU-bound), and more machines behind a load balancer (horizontal). For async apps, the event loop handles many concurrent I/O requests per worker. Add a /health endpoint so orchestrators (k8s, ECS) can probe liveness.

fastapi
# Workers = process count (vertical scaling); pick ~2*CPU
gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker

# Inside async app: concurrency is cooperative via the event loop.
# CPU-bound work blocks it — offload to a threadpool or task queue.

# Horizontal scaling: run several containers behind a load balancer.
# Use a shared store (Redis) for session state and rate limits.

# Health check endpoint for the balancer:
@app.get("/health")
def health():
    return {"status": "ok"}
20

Advanced Topics

Lifespan Events

The lifespan context manager replaces the deprecated on_event('startup'/'shutdown') handlers. Code before yield runs at startup; code after yield runs at shutdown, even if startup fails. Use it to init DB connections, ML models, and caches.

fastapi
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    # runs at startup
    print("starting up")
    app.state.cache = {}
    yield
    # runs at shutdown
    print("shutting down")
    await app.state.cache.clear()

app = FastAPI(lifespan=lifespan)

# the old @app.on_event("startup") / "shutdown" are deprecated
# in favor of the lifespan handler above.

Sub Applications (Mount)

app.mount() attaches a sub-application under a path prefix, useful for versioned APIs or combining multiple apps. Each sub-app has its own routes and OpenAPI schema; the parent's docs don't include the child's routes by default.

fastapi
from fastapi import FastAPI

app = FastAPI()

api_v1 = FastAPI()
api_v2 = FastAPI()

@api_v1.get("/items")
def v1_items():
    return {"version": 1}

@api_v2.get("/items")
def v2_items():
    return {"version": 2}

app.mount("/v1", api_v1)
app.mount("/v2", api_v2)

# GET /v1/items -> {"version": 1}
# GET /v2/items -> {"version": 2}

APIRouter

APIRouter groups related routes into a module with shared prefix, tags, dependencies, and responses. Include it via app.include_router(router) (optionally with its own prefix or tags). This is the standard pattern for structuring non-trivial apps.

fastapi
from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/items",
    tags=["items"],
    dependencies=[Depends(verify_token)],
    responses={404: {"description": "Not found"}},
)

@router.get("/")
def list_items():
    return []

@router.get("/{item_id}")
def get_item(item_id: int):
    return {"id": item_id}

# in main.py: app.include_router(router)

Static Files

StaticFiles serves a directory of static assets (images, CSS, JS) at a URL prefix. Mount it once per directory. For SPA frontends, mount a final catch-all route pointing at index.html. Static files bypass the routing system for speed.

fastapi
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles

app = FastAPI()

# serve files from ./static at /static
app.mount("/static", StaticFiles(directory="static"), name="static")

# /static/logo.png -> ./static/logo.png
# /static/css/app.css -> ./static/css/app.css

Streaming Response

StreamingResponse sends a response generated in chunks (a generator/iterator) instead of building the whole body in memory — ideal for large CSV exports, file streaming, or SSE. Set media_type appropriately so clients handle the stream correctly.

fastapi
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import csv, io

app = FastAPI()

def iter_csv():
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["id", "name"])
    for i in range(1000):
        writer.writerow([i, f"name-{i}"])
        yield output.getvalue()
        output.seek(0)
        output.truncate(0)

@app.get("/export.csv")
def export():
    return StreamingResponse(
        iter_csv(),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=export.csv"},
    )

Custom Route Class

Subclassing APIRoute lets you wrap every route handler with custom logic (logging, tracing, timing) without touching each endpoint. Set app.router.route_class or pass route_class to APIRouter. The handler runs for each matching request.

fastapi
from fastapi import FastAPI, Request, Response
from fastapi.routing import APIRoute
from typing import Callable

class LoggingRoute(APIRoute):
    def get_route_handler(self) -> Callable:
        original = super().get_route_handler()
        async def handler(request: Request) -> Response:
            print(f"-> {request.method} {request.url.path}")
            response = await original(request)
            print(f"<- {response.status_code}")
            return response
        return handler

app = FastAPI()
app.router.route_class = LoggingRoute

@app.get("/items")
def items():
    return {"items": []}

Snippets de FastAPI relacionados

Copy-paste ready code for common tasks.

Path Parameters

Capture typed path segments with validation.

Query Parameters

Parse query strings with defaults and validation.

Pydantic Request Body

Validate JSON bodies with Pydantic models.

Dependency Injection

Share logic via Depends and yield-based dependencies.

Response Model

Shape and filter responses with response_model.

JWT Auth

Issue and verify JSON Web Tokens with OAuth2.

Async and Await

Define async handlers and run blocking work in a thread.

Middleware

Add CORS, timing, and custom middleware.

Was this helpful?

Todas as folhas

Learning path

Learn from scratch

Learn this language from the ground up with structured lessons.

View lessons