Fast API
Fast API
Welcome to the definitive handbook for production-ready FastAPI backend engineering. This book is
structured into 15 phases, taking you from Python language fundamentals to advanced system design.
Chapter 3: ASGI — Web Server Gateway Interface vs ASGI, Spec Details, and Uvicorn/Gunicorn
Chapter 12: Background Tasks — FastAPI BackgroundTasks, Thread Pooling, and Basic Fire-and-
Forget Logic
Chapter 13: OpenAPI — Automatic Documentation, Swagger UI, ReDoc, Customize Schema, and
Metadata Injection
Chapter 14: Project Structure — Directory Layout, Modular Routers, Configuration Files, and
Scalable Code Architecture
Chapter 15: SQL Fundamentals & PostgreSQL in Depth — Connections, Pools, Isolation Levels,
and Multi-threading
Chapter 16: SQLAlchemy ORM — Declarative Mapping, Core vs ORM, Sessions, and Unit of Work
Pattern
Chapter 17: Async SQLAlchemy — Async Sessions, Greenlet Engine, and Concurrency in DB
Access
Chapter 18: Alembic — Migrations, Autogenerate, Version Control, and Custom Data Migration
Scripts
Chapter 19: Relationships & Joins — Lazy Loading, Eager Loading (selectin, joined), and Cascade
Behaviors
Chapter 20: Transactions & Locking — Acid Properties, Transaction Context Managers, and Row-
level Locking
Chapter 21: Indexing & Query Optimization — B-Trees, Indexes, Query Analysis (EXPLAIN
ANALYZE), and N+1 Queries
Chapter 23: Authentication Foundations & JWT — Hashing, Tokens, and Cryptographic Signatures
Chapter 24: Refresh Tokens & OAuth2 — Session Revocation, OAuth2 Flow, and Token Lifecycles
Chapter 25: Role-Based Access Control (RBAC) & Permissions — Declarative Scopes, Multi-tenant
Permissions, and Dynamic Policies
Chapter 26: User Verification Lifecycle — Email Verification, Password Reset Flows, and Secure
Token Generation
Chapter 27: File Upload Handling — UploadFile vs bytes, Streaming, and Local Storage Limits
Chapter 28: Cloud Storage Integrations — AWS S3 / MinIO Integration, Multipart Uploads, and
Signed URLs
Chapter 29: Image Processing — Pillow/PIL, Resizing, Optimization, and CDN Caching Strategies
Chapter 30: Redis Integration — Caching, Pub/Sub, and Key-Value Operations in FastAPI
Chapter 31: Celery & RabbitMQ — Task Queues, Distributed Workers, Task Results, and Retry
Policies
Chapter 33: REST API Design, Versioning, Filtering, Pagination, and Error Standards
Chapter 34: API Security & OWASP Top 10 — CORS, CSRF, XSS, SQLi, Rate Limiting, and DDOS
Prevention
Chapter 35: Testing FastAPI Apps — Pytest, Async Test Clients, Fixtures, Mocking, and Database
Isolation
Chapter 36: Docker & Docker Compose — Multi-stage Builds, Development vs Production Configs
Chapter 37: Production Web Hosting — Nginx, Uvicorn/Gunicorn tuning, Systemd, and Reverse
Proxying
Chapter 38: CI/CD Pipelines — GitHub Actions, Linters, Automating Tests, and Auto-deployment
Chapter 39: Logging & Monitoring — Structlog, OpenTelemetry, Prometheus, Grafana, and Sentry
Chapter 40: Clean Architecture & DDD — Repository Pattern, Service Layer, Unit of Work, and
SOLID Principles
Chapter 41: Advanced Performance — Async Threadpools, Connection Pooling tuning, Caching,
and Streaming Responses
Chapter 43: System Design for Backend Engineers — Real-world Case Studies: URL Shortener,
Real-time Chat, Notification Service, File Storage, and Scaled Caching
Chapter 1: Python Type Hints — Static Analysis, Runtime
Metadata, and Annotated
1. Introduction
For the first two decades of its existence, Python was known almost exclusively as a dynamically typed
language. Variable declarations did not require a type, functions could accept any object, and type
checking was performed at runtime—a philosophy often referred to as duck typing: "If it walks like a
duck and quacks like a duck, it's a duck."
While dynamic typing allowed for rapid prototyping and clean, boilerplate-free syntax, it introduced
significant challenges as applications grew in scale. Large codebases suffered from runtime crashes
due to simple type mismatches, refactoring became a high-risk gamble, and integrated development
environments (IDEs) struggled to provide accurate autocompletion and static analysis.
To bridge this gap, Python introduced type hints in PEP 484 (Python 3.5). Type hints do not change
Python's dynamic runtime execution model; instead, they provide a standardized syntax to annotate
variables, function arguments, and return types. This annotations system enables:
1. Static Analysis: External tools (such as MyPy, Pyright, and Pyre) can scan code before execution
to detect logical errors and type mismatches.
2. Enhanced Developer Experience (DX): IDEs use type annotations to offer accurate auto-
completion, inline documentation, and refactoring utilities.
3. Runtime Metadata Utilization: Frameworks like FastAPI and Pydantic inspect type hints at
runtime to automate request validation, serialization, and API documentation generation.
In FastAPI, type hints are not optional documentation details—they are the core engine of the
framework.
+-------------------------------------------------------------+
| Python Source Code |
| def register_user(age: int, email: str): |
+------------------------------+------------------------------+
|
+------------------+------------------+
| |
v v
+-----------------------+ +-----------------------+
| Static Type Checkers | | Runtime Frameworks |
| (MyPy, Pyright) | | (FastAPI / Pydantic) |
| | | |
| - Catch type bugs | | - Input Validation |
| - Enable IDE autocomplete | - Serialization |
| - Zero runtime cost | | - OpenAPI Schema |
+-----------------------+ +-----------------------+
2. Theory
Static Typing: Types are resolved and verified at compile-time (e.g., C++, Java, Rust).
Dynamic Typing: Types are resolved at runtime (e.g., Python, Ruby, JavaScript).
Strong Typing: The language prevents operations between mismatched types without explicit
conversion (e.g., Python will raise a TypeError if you try 1 + "2" ).
Weak Typing: The language performs implicit type conversions (coercion) to complete operations
(e.g., JavaScript will evaluate 1 + "2" as "12" ).
Python is strongly, dynamically typed. Type hints add gradual typing to Python, allowing you to
selectively annotate parts of your codebase, migrating incrementally from a fully dynamic style to strict
type safety.
How does a type checker decide if a type T_Sub is a valid replacement for T_Super ? There are two
primary paradigms:
2. Structural Subtyping (Static Duck Typing): Subtyping is determined by the shape (attributes and
methods) of the types. If Class B implements all methods defined in Protocol A , then B is a
subtype of A without explicitly inheriting from it. In Python, this is implemented using
[Link] .
Variance describes how subtyping of generic types (e.g., list[T] ) relates to subtyping of their
component types (e.g., T ).
Covariance: A generic type is covariant if the subtyping direction is preserved: Generic[Dog] <:
Generic[Animal] . Read-only collections (like Sequence[T] or Tuple[T, ...] ) are covariant.
3. Internal Working
When you write a type hint in Python, the Python interpreter compiles it into bytecode and stores it. It
does not enforce it. Let's look at how Python handles type hints internally.
At compile time, Python evaluates the expressions in type hints and stores them in a dictionary named
__annotations__ attached to the respective function, class, or module.
class User:
username: str
age: int
>>> User.__annotations__
{'username': <class 'str'>, 'age': <class 'int'>}
>>> greet.__annotations__
{'name': <class 'str'>, 'return': <class 'str'>}
Because type hints are evaluated when the code is imported, they must refer to objects that are already
defined in the current scope. This presents a problem for forward references—referring to a class
before it is defined.
class Node:
def set_parent(self, parent: "Node") -> None:
[Link] = parent
To resolve this system-wide, PEP 563 introduced postponed evaluation of annotations. By adding:
at the very top of a file, Python changes how it stores annotations. Instead of evaluating the expressions
and storing the resulting type objects, the compiler stores them as raw string representations:
>>> User.__annotations__
{'username': 'str', 'age': 'int'}
Because frameworks like FastAPI and Pydantic need to instantiate and validate classes at runtime, raw
string annotations are insufficient. They must resolve these strings back into real Python type objects.
They do this using typing.get_type_hints() , which evaluates the string annotations within the context
of the module's global and local namespace.
class User:
username: str
age: int
If a type refers to an undefined class or a class defined in another module without being imported,
get_type_hints() will raise a NameError at runtime. This is a common point of failure when combining
deferred annotations and complex runtime validation.
4. API Reference
Here is a comprehensive breakdown of Python's modern typing API (Python 3.10+ style).
1. [Link]
The ultimate escape hatch. When a value is typed as Any , the static type checker disables all type
checking for that value.
Warning: Avoid using Any in production. It propagates silently, neutralizing static analysis
downstream.
2. [Link]
Used to write reusable generic classes and functions that preserve type information across calls.
Syntax:
4. [Link]
Defines structural subtyping interfaces. Any class implementing the methods of a protocol is considered
a subtype of that protocol.
Syntax:
class Closeable(Protocol):
def close(self) -> None: ...
5. [Link]
Highly useful in FastAPI to define exact configurations or fixed query parameter choices.
Allows developers to attach arbitrary, framework-specific metadata to a type hint without breaking static
analyzers.
Significance: This is the foundational feature of modern FastAPI dependency injection and
validation.
Example:
5. Practical Examples
This example illustrates modern type annotation styles including unions and optional values.
# Test runs
assert parse_id(100) == 100
assert parse_id("200") == 200
assert parse_id("invalid") is None
Here, we implement a generic data repository pattern that enforces type consistency for saving and
retrieving generic objects.
class BaseRepository(Generic[T]):
def __init__(self) -> None:
self._storage: dict[int, T] = {}
self._next_id: int = 1
# Usage
class User:
def __init__(self, username: str):
[Link] = username
This example defines a structural interface for components that can export data to JSON format.
class JSONSerializable(Protocol):
def to_json(self) -> dict[str, Any]:
...
class ConfigFile:
def __init__(self, name: str, values: dict[str, Any]):
[Link] = name
[Link] = values
class SystemLog:
def __init__(self, message: str, level: str):
[Link] = message
[Link] = level
print(serialize_payload(config))
print(serialize_payload(log))
[[Link]]
python_version = "3.10"
warn_unused_configs = true
disallow_untyped_defs = true
disallow_incomplete_defs = true
check_untyped_defs = true
warn_redundant_casts = true
warn_unused_ignores = true
no_implicit_optional = true
show_error_codes = true
strict_equality = true
Always explicitly annotate the return type of your functions and endpoints—even if they return None .
When a type is broad (e.g., str | None ), use type narrowing before executing type-specific logic.
7. Common Mistakes
Mistake: Expecting Python to block invalid inputs at runtime because of a type hint.
Correction: Remember that standard Python does not run runtime checks on annotations. To validate
types dynamically at runtime, you must use a data validation framework like Pydantic or verify values
explicitly.
2. Cyclic Imports
# In user_service.py
from auth_service import AuthToken # Type import
class UserService:
def verify(self, token: AuthToken) -> bool: ...
# In auth_service.py
from user_service import UserService # Type import
class AuthService:
def authenticate(self, service: UserService) -> None: ...
# In user_service.py
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from auth_service import AuthToken
class UserService:
def verify(self, token: AuthToken) -> bool: ...
8. Performance Tips
Using postponed evaluation of annotations has a concrete performance benefit: it reduces module
import times.
Without this import, complex type definitions (such as nested generics or composite unions) are parsed
and constructed when the module is imported. With this import, they are stored as strings, avoiding
bytecode execution during start-up.
Static analysis is resolved entirely out-of-band (during development and CI/CD). Lean on compile-time
static checks, reserving runtime type inspections to initialization boundaries (e.g., parsing configurations
at startup or validating request models).
9. Security Considerations
FastAPI and Pydantic use type declarations to perform runtime data parsing and coercion. If your
endpoints lack strict type checks or fall back on broad types like Any or str , unexpected payloads can
bypass sanitization layers.
# Vulnerable Endpoint
@[Link]("/items/{item_id}")
def read_item(item_id: Any):
# If item_id is passed as a list or sql injection string,
# and passed directly to raw database operations, it poses security risks.
return database.query_by_id(item_id)
Correction: Strictly type path parameters to primitive types ( int , UUID ) or validate with Annotated
parameters to prevent malicious inputs from ever entering your route controllers.
@[Link]("/items/{item_id}")
def read_item(item_id: UUID): # Invalid format rejected at border
return database.query_by_id(item_id)
MyPy and Pyright support the reveal_type() pseudo-function. Place it in your code to verify how the
static analyzer resolves a specific variable's type.
# my_code.py
def process_data(value: int | str) -> None:
reveal_type(value) # Will print: Revealed type is 'Union[int, str]'
if isinstance(value, int):
reveal_type(value) # Will print: Revealed type is 'int'
Note: reveal_type is syntax checked by checkers but will crash if executed directly at runtime. Remove
or comment it out before deploying.
If you are developing dynamic middleware or security filters, you can inspect annotations and extract
structural metadata at runtime using typing.get_args and typing.get_origin .
EntityT = TypeVar("EntityT")
DbModelT = TypeVar("DbModelT")
Question 1: Explain the difference between nominal and structural subtyping in Python.
How does Python support both?
Answer:
Nominal subtyping is based on inheritance: Class B is a subtype of Class A if and only if B inherits
from A . Python implements nominal subtyping by default via class inheritance patterns (e.g., class
Dog(Animal) ).
Structural subtyping (also called static duck typing) defines subtyping by the presence of specific
attributes or methods. Python supports structural subtyping via [Link] . Any class that
implements all methods and variables specified in a class inheriting from Protocol is statically
considered a valid subtype, without requiring explicit subclassing.
Question 2: Why are Python mutable collections (like list[T] ) invariant, and what is
the difference between covariance and contravariance?
Answer:
Mutable collections like list[T] are invariant because allowing them to be covariant or contravariant
would violate type safety at runtime. For example, if list were covariant, a list[Dog] could be
passed to a function expecting list[Animal] . If that function appended a Cat to the list, the caller's
original list (declared to contain only Dog instances) would now contain a invalid Cat instance.
Covariance preserves subtyping relationships: If Dog is a subtype of Animal , then
ReadOnlySequence[Dog] is a subtype of ReadOnlySequence[Animal] .
Question 3: What does from __future__ import annotations do under the hood, and
how does it affect libraries like Pydantic or FastAPI?
Answer:
Under PEP 563, importing postponed evaluation of annotations changes how the compiler parses type
signatures. Instead of evaluating annotations at import time, they are stored as string literals inside
__annotations__ . This speeds up load times and resolves circular references.
However, runtime frameworks (like Pydantic and FastAPI) require resolved class references to build
validators. To reconstruct runtime classes from these strings, they call typing.get_type_hints() . If
classes reference dependencies defined elsewhere that are not imported, these checks will fail with a
NameError at runtime.
13. Exercises
Define a Logger Protocol that enforces two methods: log(msg: str) -> None and error(msg: str) -
> None . Write a function record_event(logger: Logger, message: str) -> None that leverages the
Protocol, and create two implementing classes to verify static validation behavior.
Solution:
class Logger(Protocol):
def log(self, msg: str) -> None: ...
def error(self, msg: str) -> None: ...
class ConsoleLogger:
def log(self, msg: str) -> None:
print(f"[INFO]: {msg}")
class FileLogger:
def __init__(self, filename: str):
[Link] = filename
Create a generic Cache[K, V] base class using generics. Define get(key: K) -> V | None and
set(key: K, value: V) -> None methods, ensuring keys and values are strictly typed dynamically.
Solution:
K = TypeVar("K")
V = TypeVar("V")
This mini-project is a fully typed configuration management parser. It parses settings dictionary objects
(typically populated via environment variables) and automatically converts and validates types securely.
from typing import Annotated, TypeVar, Type, Any, get_origin, get_args,
get_type_hints
class ConfigField:
"""Metadata container for custom type validations."""
def __init__(self, env_key: str, default: Any = None):
self.env_key = env_key
[Link] = default
class BaseSettings:
def __init__(self, data: dict[str, str]) -> None:
"""Parses inputs and maps types dynamically."""
# Resolve type hints at runtime (handles PEP 563 and PEP 649 automatically)
hints = get_type_hints(self.__class__, include_extras=True)
# Coerce type
try:
coerced_val = self._coerce(raw_val, base_type)
except (ValueError, TypeError) as e:
raise TypeError(f"Field '{field_name}' must be of type
{base_type.__name__}. Error: {e}") from e
# Load Settings
config = ApplicationSettings(env_mock)
# Verify Coercions
assert [Link] == "[Link]"
assert [Link] == 9000
assert config.debug_mode is True
assert config.database_url == "postgresql://user:pass@localhost:5432/db"
Key Takeaways
1. Python type hints provide gradual typing, improving code maintainability, static analysis via
MyPy/Pyright, and powering FastAPI/Pydantic runtime engines.
2. Nominal subtyping maps directly to class hierarchies, whereas structural subtyping (duck typing)
uses [Link] to define contracts based on object attributes/methods.
3. Postponed evaluation ( from __future__ import annotations ) stores type hints as string literals to
speed up imports, but libraries like FastAPI must resolve these back to actual classes using
typing.get_type_hints() .
4. [Link] is a crucial tool in modern Python web development, allowing you to attach
metadata to standard types to support dependency injection and validation.
Further Reading
1. Introduction
In traditional web backend architectures (such as Django with WSGI or Flask), concurrency is achieved
primarily through multi-threading or multi-processing. When a request arrives, the web server allocates
a thread to handle it. If that request performs an I/O operation—such as querying a database, reading a
file, or fetching data from a third-party API—the allocated thread blocks. It sits idle, waiting for the
operating system to return the data.
While a thread is blocked, it consumes operating system memory (typically 8MB stack space per thread
on Linux) and kernel overhead. Scalability is constrained by the maximum number of concurrent threads
the hardware can sustain before context-switching overhead degrades performance.
Asynchronous programming solves this bottleneck by replacing blocking wait states with cooperative
multitasking. Under this paradigm, a single thread runs an Event Loop. When an asynchronous request
performs an I/O operation, it yields control back to the event loop. The loop immediately switches to
process other tasks, returning to the original task only when the operating system signals that the
requested I/O data is ready.
This single-threaded concurrent approach allows frameworks like FastAPI (built on Starlette) to handle
tens of thousands of concurrent connections on modest hardware.
Synchronous (Thread-per-Request):
Request 1 |====== Thread 1: CPU ======|====== Thread 1: I/O (BLOCKED) ======|======
Thread 1: CPU ======|
Request 2 |====== Thread 2: CPU ======|========== Thread 2: I/O (BLOCKED)
==========|=== Thread 2: CPU ===|
Asynchronous (Event Loop, Single Thread):
|-- Req 1: CPU --|-- Req 2: CPU --|== Req 1: Wait I/O ==|== Req 2: Wait
I/O ==|-- Req 1: CPU --|
2. Theory
Concurrency is about dealing with many things at once. It is a structural property where multiple
execution flows make progress over overlapping time periods, typically by interleaving execution
(context switching).
Parallelism is about doing many things at once. It requires physical hardware execution lanes
(multiple CPU cores) running computations simultaneously.
Asynchronous programming in Python provides concurrency. It does not provide CPU parallelism.
Python's Global Interpreter Lock (GIL) enforces that only one thread executes Python bytecode at any
given instant.
Preemptive Multitasking (Threads/OS): The operating system kernel decides when a thread has
run long enough, preempts (suspends) it, and switches to another thread. The code has no control
over when this switch occurs.
Cooperative Multitasking (Async): The running code explicitly decides when to yield execution
control. It cooperates with other tasks. In Python, this boundary is declared using the await
keyword.
Coroutine Function: A function defined with async def . Calling it does not run the code; it returns
a Coroutine object.
Task: A subclass of Future that wraps a Coroutine object and schedules its execution on the event
loop.
Generators: The precursor to native coroutines. Under the hood, Python native coroutines still
utilize generator-like mechanisms ( yield / yield from ) in the interpreter, but are explicitly
structured to interface with an event loop.
3. Internal Working
To understand async Python, we must inspect the event loop mechanics and how the OS interacts with
it.
At the lowest level, native asynchronous frameworks do not block on sockets. Instead, they use non-
blocking sockets and delegate monitoring to the OS kernel. The event loop registers the file descriptors
(sockets) it is waiting on, and queries the kernel for updates using multiplexing system calls:
select / poll : Legacy mechanisms. They require scanning the entire list of descriptors, making
them inefficient for large numbers of connections.
The event loop cycles continuously, checking for completed events, dispatching callbacks, and sleeping
when no events are active.
Start Event Loop
Events Available?
Yes No
Execute Callbacks /
Sleep until Next Timer or
Suspend & Resume
Event
Coroutines
When Python encounters await coroutine_object() , the runtime suspends the execution of the
parent coroutine. It returns a marker to the event loop, effectively saying: "I am waiting on this nested
I/O. Please run other tasks until this yields a result."
The state of the suspended coroutine (local variables, instruction pointer) is preserved in memory. When
the awaited operation completes, the loop queues the coroutine back into its active run queue, resuming
execution exactly where it was suspended.
One of the most important aspects of FastAPI is its support for both asynchronous ( async def ) and
synchronous ( def ) endpoint handlers. FastAPI delegates route dispatching to Starlette. It handles
them as follows:
+-----------------------------------------+
| Incoming Request |
+--------------------+--------------------+
|
v
Is the route defined as async?
/ \
Yes No
/ \
v v
+-----------------------+ +-----------------------+
| Run Directly on the | | Offload to Starlette |
| Event Loop | | Internal Threadpool |
+-----------------------+ +-----------------------+
1. async def Routes: Starlette executes these routes directly on the event loop thread.
Crucial Rule: The code inside an async def route must never perform blocking synchronous
calls. If it blocks, it stops the event loop, freezing the entire application for all concurrent users.
2. def (Sync) Routes: Starlette offloads these to an internal Threadpool ( anyio worker threads).
The route executes in a separate thread, allowing the main thread's event loop to continue running.
Crucial Rule: Use def when interfacing with libraries that do not support async (e.g., standard
SQLAlchemy without async drivers, or synchronous file operations).
4. API Reference
Here is a reference of the core asyncio APIs used in modern production backends.
Sets up the event loop, executes the main coroutine entry point, and closes the loop upon completion.
Usage: [Link](main())
Wraps a coroutine into a Task and schedules it on the loop to run concurrently.
Returns immediately. The event loop will execute the task in the background.
Warning: If return_exceptions is False , the first raised exception will bubble up, but other
scheduled tasks will keep running in the background. Set to True to handle exceptions manually.
Executes blocking, synchronous functions in a separate thread, yielding control back to the event loop
in the main thread.
try:
await asyncio.wait_for(fetch_data(), timeout=2.0)
except [Link]:
[Link]("Query timed out")
[Link]() : Protects an awaitable from being cancelled if its parent context is cancelled.
await [Link](save_important_transaction_to_db())
5. Practical Examples
This example simulates fetching data from multiple external microservices concurrently, cutting down
the overall API response time.
import asyncio
import time
This example illustrates how to safely integrate synchronous operations (like computing a bcrypt
password hash, which blocks CPU) inside an asynchronous application.
import asyncio
import time
import hashlib
[Link](main())
Endpoint
Operation Type Core Reasoning
Definition
Replace standard open() file handlers with anyio.to_thread.run_sync or the aiofiles library.
Unbounded concurrency leads to resource exhaustion (e.g., running out of file descriptors, database
connections, or socket buffers). Always bound execution loops using an [Link] :
# Prevent system overload by capping concurrent connections to 10
semaphore = [Link](10)
7. Common Mistakes
The Bug: Using blocking synchronous functions inside an async def function.
The Fix: Use [Link] or write the route as a synchronous def function.
@[Link]("/compute")
def compute():
[Link](2) # Safe. Executed in Starlette's threadpool.
return {"status": "done"}
@[Link]("/items")
async def create_item(data: dict):
save_data(data) # Warning: Coroutine was never awaited!
return {"status": "created"}
Note: Python will log a RuntimeWarning: "coroutine 'save_data' was never awaited". The operation is
scheduled but never executed.
The Fix:
await save_data(data)
3. Leaking Task Exceptions
If you fire a task in the background using asyncio.create_task() and it crashes, the exception is
swallowed silently until the task object is collected.
The Fix: Always attach an exception-logging callback to background tasks:
task = asyncio.create_task(send_alert_email())
task.add_done_callback(handle_task_result)
8. Performance Tips
In Linux production deployments, replace the default Python event loop with uvloop . It is built on top of
libuv (the engine powering [Link]) and written in Cython, often doubling execution performance of
the event loop.
To install:
import sys
import asyncio
if [Link] != "win32":
import uvloop
asyncio.set_event_loop_policy([Link]())
9. Security Considerations
In synchronous servers, a slow client or CPU-blocking call impacts only a single worker thread. In async
applications, a single event-loop blocking call stops the entire application process.
A malicious actor could craft inputs that trigger slow mathematical processes or infinite loops in your
async def routes, bringing down the application.
Mitigation:
Protect routes with request rate limiting.
Offload all intensive CPU checks, string calculations, or hash evaluations using
asyncio.to_thread or background task queues.
Ensure all client connections and external HTTP calls have strict timeouts.
You can enable debugging to identify event loop bottlenecks. When active, asyncio will log warnings if
a coroutine blocks the loop for longer than a specified threshold (default is 100ms).
export PYTHONASYNCIODEBUG=1
Via Code:
import asyncio
loop = asyncio.get_event_loop()
loop.set_debug(True)
# Adjust warnings limit to 50 milliseconds
loop.slow_callback_duration = 0.05
Many backend services must perform health checks on multiple upstream endpoints. Querying them
sequentially is slow. Running them concurrently with custom timeouts ensures minimum response
latency.
import httpx
import asyncio
async def check_service(client: [Link], name: str, url: str) -> dict:
try:
# Award service checks a strict 3-second timeout limit
response = await [Link](url, timeout=3.0)
return {"service": name, "status": "UP" if response.status_code == 200 else
"DOWN"}
except Exception as e:
return {"service": name, "status": f"ERROR: {str(e)}"}
Question 1: How does FastAPI run synchronous def routes concurrently? Under what
scenario would they block?
Answer:
FastAPI executes def routes by offloading them to Starlette's threadpool ( anyio worker threads). The
main thread running the event loop is not blocked because the synchronous code executes on a
separate thread.
However, thread pools have a finite capacity (e.g., default limit of 40 threads). If your server receives
hundreds of concurrent requests targeting blocking def endpoints, all thread pool lanes will be
occupied. Any subsequent request will queue up, blocking the application's ability to handle those routes
until threads free up.
Question 2: Why is running [Link]() inside an async def route a critical mistake
in FastAPI? How do you resolve it?
Answer:
[Link]() is a synchronous, blocking HTTP call. Because it is executed inside an async def
function, it runs directly on the main event loop thread. While it waits for the server response, the event
loop thread is blocked. No other coroutines or requests can make progress, resulting in severe latency
spikes or timeouts for all concurrent users.
To resolve this, you should either:
2. Redefine the route as a synchronous def function so it executes inside the thread pool.
Answer:
[Link]() accepts a list of awaitables, runs them concurrently, and returns the
accumulated results in the exact order they were requested. It is designed for aggregating data.
[Link]() takes a set of Task objects and returns two sets: (done, pending) . It allows for
more granular control over lifecycles, letting you return when the first task finishes
( return_when=FIRST_COMPLETED ), handles cancellation, and requires manual result unpacking.
13. Exercises
Write a function download_batch(urls: list[str]) -> list[str | None] that retrieves text content
from a list of URLs concurrently using [Link] . Set a timeout of 2 seconds for each
download, and return None for any URL that fails.
Solution:
import asyncio
import httpx
This mini-project implements a highly concurrent local file crawler. It recursively traverses a folder, reads
metadata from all text files in parallel, caps concurrent execution using a Semaphore to protect file
system handle limits, and aggregates findings.
import os
import asyncio
import time
from pathlib import Path
class FileMetadataCrawler:
def __init__(self, max_concurrent_reads: int = 5):
# Cap concurrent file reads to protect OS file descriptors
[Link] = [Link](max_concurrent_reads)
return {
"file_name": file_path.name,
"path": str(file_path),
"size_bytes": stats.st_size,
"last_modified": stats.st_mtime,
"preview": first_line
}
if not target_files:
return []
crawler = FileMetadataCrawler(max_concurrent_reads=3)
start_time = time.perf_counter()
try:
metadata_records = await crawler.crawl_directory(target_dir)
elapsed = time.perf_counter() - start_time
if __name__ == "__main__":
[Link](run_crawler())
Key Takeaways
1. Asynchronous programming provides cooperative multitasking via an Event Loop, enabling a single
thread to handle highly concurrent I/O operations without blocking.
2. The operating system handles socket pooling behind the scenes using high-performance
multiplexing syscalls like epoll (Linux) and kqueue (macOS).
3. FastAPI (Starlette) runs async def endpoints on the event loop, and runs synchronous def
endpoints in a separate worker threadpool.
4. Blocking the event loop with synchronous calls (like [Link] or blocking I/O) stops execution for
all concurrent requests and must be avoided.
Further Reading
1. Introduction
To understand how modern Python web frameworks handle requests, we must look at the interfaces
between web servers and application code. Historically, Python relied on WSGI (Web Server Gateway
Interface), specified in PEP 3333.
WSGI was a massive success, standardizing how synchronous frameworks (like Django and Flask)
communicated with web servers (like Gunicorn or uWSGI). However, WSGI is strictly synchronous. It
assumes a simple model: a request comes in, a function is called, it processes the request
synchronously, and it returns a response.
To bridge this gap, ASGI (Asynchronous Server Gateway Interface) was created. ASGI is a spiritual
successor to WSGI, designed to support common asynchronous protocols like HTTP, HTTP/2, and
WebSockets.
The differences between WSGI and ASGI lie in their signature definitions and concurrency capabilities:
WSGI Signature:
WSGI is a single synchronous call. The app cannot process incoming data after returning the
iterable, nor can it handle long-polling connections or WebSockets without blocking the worker
thread.
ASGI Signature:
async def asgi_app(scope: dict, receive: Callable, send: Callable) -> None:
...
ASGI is an asynchronous function that takes a connection scope and two asynchronous streams:
receive (to read incoming data) and send (to write outgoing data).
1. Connection Lifecycle: Created when a client connects (e.g., an HTTP request or a WebSocket
handshake) and destroyed when the connection closes.
2. Lifespan Lifecycle: Spans the entire lifetime of the web server process. The Lifespan Protocol
allows applications to register startup and shutdown hooks (e.g., initializing database connection
pools on startup and cleaning them up on shutdown).
3. Internal Working
scope : A dictionary containing metadata about the connection. It acts as the context. For an HTTP
request, it details the HTTP version, request method, path, headers, query parameters, client IP,
and scheme.
receive : An async function that the application calls to receive incoming events from the server.
For example, reading chunks of a large file upload payload ( [Link] events).
send : An async function that the application calls to send events back to the client. For example,
starting the response headers ( [Link] ) and streaming the body bytes
( [Link] ).
1. Uvicorn parses the raw TCP packets, extracts HTTP headers, and initializes the scope dictionary.
2. Uvicorn invokes the ASGI application, passing the scope , along with references to its internal
receive and send channels.
3. The ASGI application processes the request. If it needs to read the request body, it await
receive() .
5. Once the application function completes, Uvicorn cleans up the socket connection.
4. API Reference
ASGI uses structured dictionaries (events) for communication. Let's inspect the core events defined by
the ASGI HTTP spec:
1. Incoming: [Link]
Fields:
more_body (boolean): If True , indicates the client is streaming more body chunks.
2. Outgoing: [Link]
Fields:
3. Outgoing: [Link]
Fields:
5. Practical Examples
This is a raw ASGI application that runs without FastAPI or Starlette. It parses the request path, extracts
a custom header, and returns a dynamic HTML page.
import json
if path == "/":
status = 200
response_data = b"<h1>Welcome to the Bare ASGI Homepage</h1>"
content_type = b"text/html"
elif path == "/api/status":
status = 200
response_data = [Link]({"status": "healthy", "engine":
"raw_asgi"}).encode()
content_type = b"application/json"
else:
status = 404
response_data = b"Not Found"
content_type = b"text/plain"
[Internet Client] ---> [Nginx (Reverse Proxy / SSL)] ---> [Uvicorn (ASGI Server)]
Nginx acts as the front-line shield. It handles SSL termination, buffers slow client uploads, serves
static files directly, and protects against simple denial-of-service threats.
Gunicorn acts as the process manager. It manages multiple worker processes, restarts crashed
workers, and controls process limits.
7. Common Mistakes
# Antipattern: Crash!
async def app(scope, receive, send):
await send({
"type": "[Link]",
"body": b"Hello World"
})
The Fix: You must send the response start headers first so the server can construct the HTTP frame.
Just like the event loop inside standard async routes, the ASGI server loop is single-threaded per
worker. If you run blocking DB queries (such as synchronous psycopg2 calls) directly inside a raw ASGI
middleware or handler, the server cannot accept other connections.
The Fix: Offload blocking I/O calls to worker threads using anyio.to_thread or standard executor
threads.
8. Performance Tips
Ensure that your ASGI server is configured to utilize HTTP Keep-Alive. Keeping TCP sockets open for
repeat requests reduces handshake overhead:
Tuning --backlog defines the maximum length of the queue of pending connections. Setting this
higher (e.g., 2048 ) helps prevent dropped connections under high load.
9. Security Considerations
At the bare ASGI layer, it is crucial to protect your server from payload crashes (e.g., a client uploading
a 10GB file directly to memory). Implement a size-limiting check in ASGI middleware:
bytes_received = 0
When Uvicorn runs behind Nginx, it does not see the real client IP; it sees Nginx's IP ( [Link] ).
Ensure Nginx passes the real IP headers (e.g., X-Forwarded-For ), and configure Uvicorn with --proxy-
headers and --forwarded-allow-ips to prevent IP-spoofing attacks.
To debug path issues, header parsing errors, or missing cookies, write a simple debugging decorator
that logs the complete scope structure for every request:
import pprint
class ASGIDebugMiddleware:
def __init__(self, app):
[Link] = app
A common requirement is to block malicious IPs before they reach the router or execute heavy
database code. Doing this at the ASGI entry point maximizes performance.
class IPBlacklistMiddleware:
def __init__(self, app, blocked_ips: set[str]):
[Link] = app
self.blocked_ips = blocked_ips
Question 1: What is the core difference between scope , receive , and send in the ASGI
spec?
Answer:
scope is a dictionary created once at the beginning of the connection containing metadata (e.g.
headers, path, method). It is read-only and static.
receive is an asynchronous callable that allows the application to pull event messages (like
request body chunks or WebSocket disconnects) from the server.
send is an asynchronous callable that allows the application to push event messages (like
response headers or response body chunks) back to the server.
Question 2: Explain how the ASGI Lifespan protocol works and why it is critical for
database-backed web services.
Answer:
The Lifespan protocol allows the web server to communicate process-wide lifecycles to the application.
When the server starts up, it sends a [Link] event. The application catches this, executes
startup tasks (like establishing database connection pools, cache clients, or model loading), and
responds.
When shutting down, the server sends a [Link] event, allowing the application to close
connections cleanly, write logs, and release resources. Without it, databases could suffer from dangling
open connections or data leaks on abrupt process restarts.
13. Exercises
Write an ASGI middleware that intercepts all outgoing HTTP responses and injects a custom header: X-
Security-Policy: High-Strictness .
Solution:
class HeaderInjectionMiddleware:
def __init__(self, app):
[Link] = app
This mini-project is a fully functional static file server written directly against the raw ASGI interface. It
handles async file streaming, returns correct MIME types, and responds with a 404 Not Found if a file
does not exist.
import os
import mimetypes
from pathlib import Path
class BareStaticFileServer:
def __init__(self, root_directory: str):
self.root_path = Path(root_directory).resolve()
if not self.root_path.exists():
raise FileNotFoundError(f"Static directory {root_directory} does not
exist.")
method = scope["method"]
if method not in ("GET", "HEAD"):
await self.send_error(send, 405, "Method Not Allowed")
return
# Start response
await send({
"type": "[Link]",
"status": 200,
"headers": [
[b"content-type", content_type.encode()],
[b"content-length", str(file_size).encode()],
[b"server", b"bare-static-server"]
]
})
if method == "HEAD":
# HEAD requests only get headers
await send({
"type": "[Link]",
"body": b"",
"more_body": False
})
return
await send({
"type": "[Link]",
"body": chunk,
"more_body": (bytes_sent < file_size)
})
except Exception as e:
# File system error fallback
pass
async def send_error(self, send, status_code: int, message: str) -> None:
await send({
"type": "[Link]",
"status": status_code,
"headers": [[b"content-type", b"text/plain"]]
})
await send({
"type": "[Link]",
"body": [Link](),
"more_body": False
})
Key Takeaways
1. WSGI is a legacy synchronous interface, while ASGI is designed to handle asynchronous routing,
streaming HTTP payloads, and WebSockets.
3. Running raw ASGI middleware allows you to intercept connections early, blocking bad IPs, logging
inputs, or serving static assets before invoking high-level frameworks.
4. Gunicorn behaves as a production manager overseeing worker processes, which run local Uvicorn
event loop engines.
Further Reading
1. Introduction
FastAPI is often described as a web framework, but under the hood, it is an orchestration layer. It does
not reinvent the wheel for web servers, routing, or data validation. Instead, it sits on the shoulders of two
core libraries:
1. Starlette: A lightweight ASGI framework that provides routing, middleware, state management, and
request/response structures.
2. Pydantic: A data validation and serialization library that enforces type safety using Python type
hints.
FastAPI acts as the glue, combining Starlette's high-performance asynchronous web routing engine with
Pydantic's data validation capabilities. It adds:
2. Theory
To build a clean architecture, we must understand which components handle which parts of the request
lifecycle:
Starlette's Responsibilities:
Routing: Inspecting incoming request paths and matching them to target endpoints.
Protocols: Managing the raw HTTP request parsing and WebSocket handshake operations.
Middleware: Running cross-cutting code (CORS, trusted hosts, authentication) on requests and
responses.
Exceptions: Catching general server errors and returning basic error responses.
Pydantic's Responsibilities:
Parsing & Coercion: Converting incoming string parameters (e.g., query params, headers,
form data) or JSON bodies into validated Python types.
Serialization: Dumping complex Python objects (like database models) into clean JSON-
serializable dictionaries.
Schema Definition: Providing metadata that defines the structural layout of API endpoints.
FastAPI's Responsibilities:
Compilation: Inspecting route signature parameters ( def endpoint(name: str) ) and compiling
them into parameter resolvers.
3. Internal Working
When a FastAPI application starts up, it does not wait for requests to inspect type signatures. It
performs signature compilation during module import and initialization.
Start App --> Inspect route functions via `[Link]` --> Map parameters to
fields --> Generate Pydantic Models for Inputs --> Register Starlette Route
1. Parameter Inspection: FastAPI uses Python's inspect module to parse the signature of your
route function.
2. Dynamic Pydantic Models: For every endpoint, FastAPI dynamically builds a Pydantic model
representation of the expected inputs (query parameters, path variables, request bodies).
3. Route Registration: FastAPI wraps the route function inside an internal Starlette Route object,
linking it to the dynamically generated input parser and validation schemas.
1. Starlette Router: Matches the request URL path to the registered route.
3. Validation: FastAPI feeds these parameters into the dynamically constructed Pydantic input model.
If validation fails, it catches the Pydantic ValidationError and translates it into a Starlette-
compatible HTTPException with a 422 Unprocessable Entity status code and structured
validation details.
4. Dependency Resolution: FastAPI resolves and executes the endpoint's dependency graph.
5. Execution: The endpoint function is called with the resolved parameters and injected
dependencies.
6. Serialization: The returned value is run through Pydantic's serializer to convert it to a standard
JSON-compatible Python structure.
7. Starlette Response: Starlette wraps the serialized data into an HTTPResponse and streams it back
to the client.
4. API Reference
Extensibility Hooks
Custom APIRouter subclassing
FastAPI provides the APIRouter class to group operations together. If you need to enforce custom
behavior (e.g., logging every endpoint call or checking permissions), you can define a custom routing
class.
class CustomAPIRoute(APIRoute):
def get_route_handler(self) -> Callable:
original_handler = super().get_route_handler()
return custom_handler
5. Practical Examples
This example demonstrates how Starlette and Pydantic operate in unison within a standard FastAPI
application.
Avoid establishing new database connection pools or network client sessions (like [Link] )
inside route functions. Creating connections is expensive and causes socket exhaustion.
Initialize them once during startup inside the lifespan event handler, and bind them to the application
state context.
class DBConnectionPool:
async def close(self): pass
@asynccontextmanager
async def lifespan(app: FastAPI):
# Startup: Initialize shared pools
[Link].db_pool = DBConnectionPool()
[Link].http_client = [Link]()
yield
# Shutdown: Clean up resources
await [Link].db_pool.close()
await [Link].http_client.aclose()
app = FastAPI(lifespan=lifespan)
7. Common Mistakes
The Bug: Registering routers within conditional code paths or function invocations during runtime.
Routers must be registered at the module-import level. Since FastAPI compiles validation metadata at
start-up, registering routers dynamically during request execution will bypass signature compilation,
resulting in runtime AttributeError or unvalidated routes.
The Fix: Keep route definitions static and modularized in clean sub-router files.
Starlette allows storing global variables in [Link] . Mutating state fields inside asynchronous request
handlers without a concurrency lock can cause race conditions.
The Fix: Treat [Link] attributes as read-only pools populated during the startup phase.
8. Performance Tips
If an endpoint returns huge dataset responses (e.g. returning a 50MB raw list of dictionaries queried
from database), FastAPI's automatic Pydantic serialization can introduce high CPU overhead.
If your data is already formatted as a JSON-serializable list/dictionary, you can bypass Pydantic
serialization by returning a Starlette JSONResponse directly:
@[Link]("/large-dataset")
async def get_large_dataset():
data = db.query_raw_dictionaries()
# Fast native serialization, skipping Pydantic mapping
return JSONResponse(content=data)
9. Security Considerations
By default, FastAPI exposes Swagger ( /docs ) and ReDoc ( /redoc ) open-access points. In production
environments, exposing these routes allows malicious actors to map your API attack surface.
app = FastAPI(
docs_url=None if [Link]("ENV") == "prod" else "/docs",
redoc_url=None if [Link]("ENV") == "prod" else "/redoc",
openapi_url=None if [Link]("ENV") == "prod" else "/[Link]"
)
If routes are not responding, or you suspect path overlaps, inspect the application's compiled routing
list. You can print the route paths and methods programmatically:
@app.on_event("startup")
def print_routes():
for route in [Link]:
# Starlette Route objects expose name, path, and methods
methods = getattr(route, "methods", None)
print(f"Path: {[Link]} | Methods: {methods} | Name: {[Link]}")
Many enterprise applications require dynamically injecting custom rate-limiting metadata or routing logic
to routes. By modifying the underlying APIRoute class, we can create a clean decorator annotation
pattern.
class RateLimitRoute(APIRoute):
def get_route_handler(self):
original_handler = super().get_route_handler()
Question 1: How does FastAPI translate a Pydantic Validation Error to an HTTP 422
response?
Answer:
When a client sends an invalid payload, the Starlette router passes control to FastAPI's validation layer.
FastAPI feeds the input to Pydantic, which raises a [Link] . FastAPI's internal
exception handler catches this error, extracts the validation logs ( [Link]() ), and wraps them into
a structured RequestValidationError object.
FastAPI has a default exception handler registered for RequestValidationError . This handler formats
the errors into an array of error messages (detailing locate, message, and type) and returns them to
Starlette as a JSONResponse with a 422 Unprocessable Entity status code.
Question 2: Why is the separation of Starlette and Pydantic important for FastAPI's
design?
Answer:
This separation of concerns follows clean software engineering principles. Starlette focus is handling
asynchronous HTTP networking, connection lifecycles, and routing safely. Pydantic's focus is validating,
coercing, and documenting data schemas based on type annotations.
FastAPI acts as the clean integration layer. This architecture allows developers to swap components
(e.g. using custom response serializers) or extend validation metadata independently, without coupling
web protocol layers with data validations.
13. Exercises
Implement a subclass of Starlette's APIRoute that measures the time elapsed during endpoint
processing, injecting an X-Response-Time-Ms response header automatically.
Solution:
import time
from [Link] import APIRoute
from fastapi import Request, Response
class TimingRoute(APIRoute):
def get_route_handler(self):
original_handler = super().get_route_handler()
async def custom_handler(request: Request) -> Response:
start_time = time.perf_counter()
response: Response = await original_handler(request)
duration = (time.perf_counter() - start_time) * 1000
[Link]["X-Response-Time-Ms"] = f"{duration:.2f}"
return response
return custom_handler
To understand how FastAPI orchestrates Starlette and Pydantic, we will build a minimal framework
clone called MiniFastAPI from scratch. It uses Starlette's Router for web requests and Pydantic for
request body validation.
class MiniFastAPI:
def __init__(self):
# Initialize the underlying Starlette router instance
self.starlette_app = Starlette()
Key Takeaways
1. FastAPI acts as an integration manager, orchestrating Starlette for async web routing and Pydantic
for type validation and serialization.
2. Route compilation occurs at import time, where FastAPI inspects function signatures and
dynamically generates validation classes to handle query/path/body inputs.
3. Returning standard Starlette responses like JSONResponse or file streams bypasses Pydantic's
serialization layer, which is useful for optimizing large payload endpoints.
4. Exposing public documentation engines in production environments exposes your API architecture
to security threats and should be disabled.
Further Reading
1. Introduction
In web development, routing is the process of mapping incoming HTTP requests (identified by their
URL path and HTTP method) to specific handler functions (controllers). A robust router must resolve
dynamic URLs (e.g. /users/100 ), validate parameters, handle query strings, and respond with
appropriate error codes (such as 404 Not Found or 405 Method Not Allowed ) when matches fail.
In legacy frameworks, routes were often defined in a centralized routing file or required configuring
complex regular expressions. FastAPI inherits Starlette's high-performance routing table and wraps it in
a pythonic, decorator-based interface.
Furthermore, FastAPI's router integrates with your parameter declarations. It analyzes your route
signature variables at start-up, determines if a parameter belongs in the URL path, the query string, or
the request body, and configures the input parsing and validation rules automatically.
2. Theory
GET: Retrieve a resource. Safe and idempotent (must not alter system state).
Path Parameters: Embedded in the URL path itself (e.g., /users/{user_id} ). They identify a
specific resource.
Query Parameters: Key-value pairs appended after the ? in the URL (e.g., /search?
query=fastapi&limit=10 ). They are typically optional and used for filtering, sorting, or pagination.
3. Internal Working
When you define a dynamic path like /items/{item_id} , FastAPI compiles it into a regular expression
at startup.
Under the hood, Starlette replaces {item_id} with a capture group regex: (?P<item_id>[^/]+) .
If you add a type converter (like {item_id:int} ), Starlette uses a specific integer-matching regex: (?
P<item_id>\d+) . This ensures that a request to /items/abc fails to match the route, immediately
returning a 404 Not Found or allowing subsequent routes to match, rather than hitting the endpoint and
failing during type coercion.
When a request arrives, Starlette's router walks down the registered routes list sequentially, testing
matches. It executes the first route that matches. This makes route definition order critical.
@[Link]("/users/me")
async def get_current_user():
return {"user": "current"}
If a client requests /users/me , the router checks the first route: /users/{user_id} . Since "me" is a
valid string, it matches. The router calls get_user(user_id="me") , and the second endpoint is never
reached.
4. API Reference
3. Headers: Header(...)
Extracts values from request headers. By default, FastAPI automatically converts underscore
characters ( _ ) to hyphens ( - ).
4. Cookies: Cookie(...)
5. Practical Examples
In production applications, do not attach all routes to the main app = FastAPI() instance. Instead,
partition your API into separate modules using APIRouter , and register them using
app.include_router() .
app/
├── [Link]
├── api/
│ ├── v1/
│ │ ├── endpoints/
│ │ │ ├── [Link]
│ │ │ └── [Link]
│ │ └── [Link]
Inside app/api/v1/endpoints/[Link] :
@[Link]("/")
async def list_users():
return []
api_router = APIRouter()
api_router.include_router([Link], prefix="/users", tags=["Users"])
api_router.include_router([Link], prefix="/items", tags=["Items"])
Inside app/[Link] :
app = FastAPI()
app.include_router(api_router, prefix="/api/v1")
7. Common Mistakes
Mistake: Declaring two routes with overlapping path configurations in the wrong order (as shown in the
internal working section).
Correction: Place all static routes at the top of the router file, followed by dynamic pattern-matching
routes.
While APIRouter allows nesting routers within other routers, keep the nesting tree relatively flat.
Every incoming request must be evaluated against the compiled routing tree. An excessively deep or
complex routing structure with many regex variables adds overhead during path parsing.
9. Security Considerations
If you write an endpoint that accepts a file path parameter and reads from the filesystem, a client can
send directory traversal vectors ( ../ ) to access restricted server files:
# VULNERABLE ENDPOINT!
@[Link]("/files/{file_path:path}")
async def get_file(file_path: str):
# A request to /files/../../etc/passwd bypasses isolation
return open(f"/var/www/static/{file_path}").read()
Correction: Never trust raw path parameters for filesystem operations. Always resolve and verify that
the target path remains within your app's isolated folder boundaries.
You can write a simple startup check to print the entire routing map of your application to verify path
layouts and detect overlapping routes:
def inspect_routing_table(app):
print("=== Compiled Routing Table ===")
for route in [Link]:
methods = ",".join([Link]) if getattr(route, "methods", None) else
"LIFESPAN"
print(f"[{methods:^10}] -> Path: {[Link]:<40} (Handler:
{[Link].__name__})")
In multi-tenant SaaS platforms, you often need to route requests dynamically based on subdomains
(e.g. [Link] vs [Link] ). This is best handled by extracting subdomain details
inside a middleware, appending it to the request state context, and referencing it inside routes.
Question 1: What is the difference between path parameters and query parameters in
terms of route resolution in FastAPI?
Answer:
Path Parameters are part of the URL structure itself. They are matched by Starlette's router using
regex capture groups compiled at startup. If a path parameter is missing or violates path converters
(such as sending a string to {id:int} ), the request fails to match the route, resulting in a 404 Not
Found .
Query Parameters are optional key-value pairs appended to the query string after the ? . They are
not evaluated during Starlette path matching. Instead, the router matches the route path first, then
delegates parsing of the query string to FastAPI's validation layer. If query validation fails, the server
returns a 422 Unprocessable Entity response.
Question 2: Why are route parameters case-insensitive and how can we customize
header extraction casing?
Answer:
Route parameters are processed as parsed strings from the URL. However, HTTP header names are
case-insensitive according to the HTTP specification. FastAPI's Header utility automatically normalizes
header names by default (converting lowercase inputs and mapping underscores to hyphens). If you
need to read a header exactly as-is without casing normalization, set convert_underscores=False .
13. Exercises
Create a route /products that accepts two query parameters: offset (must be >= 0, defaults to 0) and
limit (must be between 1 and 50, defaults to 10). Return a dictionary confirming the query limits.
Solution:
@[Link]("/products")
async def get_products(
offset: int = Query(default=0, ge=0),
limit: int = Query(default=10, ge=1, le=50)
):
return {"pagination": {"offset": offset, "limit": limit}}
This mini-project implements a modular bookstore catalog API. It demonstrates nested routers, path
validation rules, and structured OpenAPI grouping.
AUTHORS_DB = {
1: {"name": "Author Alice", "country": "US"},
2: {"name": "Author Bob", "country": "UK"}
}
@authors_router.get("/{author_id}")
async def get_author(author_id: int = Path(..., gt=0)):
author = AUTHORS_DB.get(author_id)
if not author:
raise HTTPException(status_code=404, detail="Author not found")
return author
@books_router.get("/")
async def list_books(
genre: str | None = Query(default=None, description="Filter books by genre"),
limit: int = Query(default=10, ge=1, le=20)
):
books = list(BOOKS_DB.values())
if genre:
books = [b for b in books if b["genre"].lower() == [Link]()]
return books[:limit]
@books_router.get("/{book_id}")
async def get_book(book_id: int = Path(..., description="The ID of the book",
gt=0)):
book = BOOKS_DB.get(book_id)
if not book:
raise HTTPException(status_code=404, detail="Book not found")
return book
@books_router.post("/", status_code=201)
async def add_book(payload: BookCreate):
# Verify author exists first
if payload.author_id not in AUTHORS_DB:
raise HTTPException(status_code=400, detail="Invalid author_id")
new_id = max(BOOKS_DB.keys()) + 1
BOOKS_DB[new_id] = payload.model_dump()
return {"id": new_id, "book": BOOKS_DB[new_id]}
1. Routing maps incoming requests to endpoint handlers. In FastAPI, these mappings are compiled
into high-performance regular expressions during startup.
2. Route definitions must follow a strict ordering rule: static paths must always be registered before
dynamic, parameter-matching path parameters.
3. Path parameters are used for resource identification and are evaluated during route matching.
Query parameters are used for optional modifications like filtering or pagination.
4. Modular routing structures using APIRouter keep codebases maintainable, self-contained, and
clean as they scale.
Further Reading
1. Introduction
When building a web application, different parts of your code need to share resources. For example, a
route handler that registers a user needs access to a database connection, a hashing utility, and maybe
an email client.
In a naive codebase, you might create these resources directly inside your function:
@[Link]("/register")
async def register(user: UserSchema):
db = DatabaseConnection() # Creating resource inside function
hasher = PasswordHasher()
# ... process registration
1. Tight Coupling: Your route handler is now tightly coupled to DatabaseConnection . If you change
how database connections are created, you have to modify every route handler.
2. Difficult Testing: You cannot easily test this function with a mock database. The function creates its
own database connection, making isolation impossible.
3. Resource Management: It is hard to manage the lifecycle of resources. You might leak connections
if you forget to close them.
Dependency Injection (DI) solves this by reversing control. Instead of the function creating its
dependencies, the dependencies are injected (passed in) by an external system.
FastAPI features a built-in dependency injection system that is extremely powerful, type-safe, and easy
to use.
2. Theory
What is a Dependency?
A dependency is simply an object, service, or value that your code needs to perform its job.
In FastAPI, a dependency is defined as a callable (a function, class, or callable object) that returns a
value.
When an endpoint calls a dependency that calls another dependency, FastAPI builds a Directed
Acyclic Graph (DAG) of dependencies.
For example:
FastAPI resolves this graph automatically. It runs the dependencies in the correct order, shares results
where appropriate, and passes the final resolved values into your route handler.
[Route Endpoint]
/ \
v v
[get_token] [get_db_session]
\ /
v v
[get_settings]
3. Internal Working
Request-Scoped Lifecycle
4. Once the response is sent, the request scope ends, and the cached resources are cleaned up.
For dependencies that need setup and teardown steps (like opening a database session and closing it),
Python generators ( yield ) are used.
FastAPI runs the code before the yield statement, passes the yielded value to the endpoint, lets the
endpoint execute, and then resumes execution after the yield statement to execute cleanup logic.
4. API Reference
If you need a dependency to execute multiple times instead of reusing the cached result within a
request, set use_cache=False .
3. Class-Based Dependencies
Instead of functions, you can use classes as dependencies. The class constructor is called, and the
class instance is injected.
Example:
class CommonParams:
def __init__(self, limit: int = 10, offset: int = 0):
[Link] = limit
[Link] = offset
# Usage:
params: CommonParams = Depends(CommonParams)
(Note: Because of Python's type hints, you can write Depends() without arguments, and FastAPI
will automatically default to the class type annotation).
5. Practical Examples
This example shows a database session dependency and a current user validator dependency working
together.
app = FastAPI()
Declare security dependencies (OAuth2 scopes, API key validations) in centralized dependencies.
Reuse them across routers to ensure consistent authentication checks across your endpoints.
7. Common Mistakes
Mistake: Using blocking operations inside a yield generator without wrapping it in a thread executor.
Correction: Either use an async driver, define the dependency function as a synchronous def (so
FastAPI runs it in a background thread), or wrap the cleanup block in anyio.to_thread.run_sync .
If a route handler raises an error (e.g., an HTTPException ), that error will bubble up through the yield
statement. If you trap all exceptions in a broad try/except block, you might unintentionally catch and
hide application errors.
Correction: Only catch exceptions related to connection teardown, or re-raise errors:
try:
yield db
except Exception:
# Do database rollback
raise
finally:
[Link]()
8. Performance Tips
FastAPI caches dependency results within a request by default. If your dependency performs light
operations (like reading JWT headers), caching is great.
If a dependency yields a dynamic cursor or stream, or does not need sharing, set use_cache=False to
prevent unnecessary cache lookup overhead and memory pinning.
9. Security Considerations
FastAPI's dependency injection system executes before your endpoint code runs. This makes
dependencies the ideal boundary for security enforcement (e.g., checks for authentication tokens, rate
limits, or user roles).
If you encounter circular dependencies (e.g., Dependency A depends on B, which depends on A),
FastAPI will raise a RuntimeError at startup.
To debug, map out your dependencies visually using a simple graph or review your imports to locate the
cycle.
11. Real-world Use Cases
In a large application, you can inject settings or feature flags dynamically. This allows you to toggle
feature gates or swap out active credentials without modifying application core logic.
class Settings:
def __init__(self):
self.enable_v2_features = True
@[Link]("/features")
async def list_features(settings: Annotated[Settings, Depends(get_settings)]):
return {"v2_active": settings.enable_v2_features}
Question 1: How does Dependency Caching work in FastAPI, and how do you disable
it?
Answer:
Within a single HTTP request context, FastAPI compiles a graph of dependencies. If the same
dependency is declared in multiple locations in the graph (e.g., both a user dependency and an items
dependency need a database session), FastAPI resolves it once, caches the result, and injects that
same cached instance into subsequent references.
You can disable caching by setting use_cache=False when declaring the dependency:
Depends(get_db, use_cache=False) .
Answer:
If a route handler raises an exception, the exception is raised at the location of the yield statement in
the dependency. This allows the dependency to catch the exception, perform rollbacks or cleanup, and
then let the exception propagate. It is critical to ensure that any try/finally block inside the
dependency allows the exception to bubble up so that FastAPI's exception handlers can return the
correct error response to the client.
13. Exercises
Write a dependency verify_api_key that checks for a header named X-API-Key . If the header value is
not secret-handshake , raise an HTTPException with a 403 Forbidden status code.
Solution:
14. Mini Project: Configurable File Database and Testing Override Mock
This mini-project demonstrates how to implement a clean database service using dependencies and
override it during testing to ensure zero file system side effects.
import os
from typing import Annotated
from fastapi import FastAPI, Depends, HTTPException
# 2. Dependency Provider
def get_file_db() -> LocalFileDatabase:
# In production, returns access to local [Link]
return LocalFileDatabase("[Link]")
# 3. Application setup
app = FastAPI()
@[Link]("/records")
async def read_records(db: Annotated[LocalFileDatabase, Depends(get_file_db)]):
return {"records": db.read_records()}
@[Link]("/records")
async def add_record(
record: str,
db: Annotated[LocalFileDatabase, Depends(get_file_db)]
):
db.write_record(record)
return {"status": "success"}
class MockFileDatabase:
"""Mock database that stores records in memory to prevent writing files."""
def __init__(self):
[Link] = []
# Verification:
# When endpoints are called now, they will receive mock_db instead of
LocalFileDatabase.
Key Takeaways
1. Dependency Injection decouples your business logic from resource management, making
codebases easier to maintain and test.
3. Yield-based dependencies are used to manage resource lifecycles (like database connections),
executing setup code before yielding and cleanup code after the response is sent.
Further Reading
1. Introduction
When building a web API, you cannot trust data sent by clients. Users can send malformed JSON,
strings instead of numbers, database-breaking injection payloads, or missing fields.
Runtime Crashes: A function expecting a number crashes when trying to add a string.
Database Corruption: Storing garbage values that break reporting or downstream consumer
systems.
Input Validation is the practice of checking incoming data at the application boundary (the API
endpoints) and rejecting payloads that do not meet strict specifications.
In FastAPI, input validation is handled by Pydantic. You define the structure and constraints of your
data using standard Python type hints, and FastAPI verifies incoming data automatically, returning a
detailed 422 Unprocessable Entity response to the client if the data is invalid.
2. Theory
By default, Pydantic uses smart type coercion. It tries to convert incoming values to the declared
target types if it can do so without losing data.
If a field is typed as int and the client sends "123" (string), Pydantic will convert it to the integer
123 .
If a field is typed as bool and the client sends "true" , "yes" , or 1 , Pydantic will convert it to
True .
If Pydantic cannot safely convert a value (e.g., trying to parse "abc" as an int ), it rejects the request.
If you prefer to disable this behavior and reject values that do not match the exact type, Pydantic
supports strict mode.
Input Sanitization
Validation verifies that the data fits a structure. Sanitization cleans the data to make it safe for
processing (e.g., stripping HTML tags to prevent XSS, or trim spaces from usernames).
3. Internal Working
2. FastAPI intercepts this error at the routing layer and converts it into a
[Link] .
3. FastAPI's built-in exception handler formats the error list into a clean JSON structure:
{
"detail": [
{
"loc": ["body", "age"],
"msg": "Input should be a valid integer",
"type": "int_parsing"
}
]
}
4. It sends this payload to the client with a 422 Unprocessable Entity HTTP status code, preventing
the endpoint function from ever running.
4. API Reference
Common Constraints:
Syntax:
@field_validator("field_name")
@classmethod
def check_value(cls, value: Any) -> Any:
if not is_valid(value):
raise ValueError("Custom error message")
return value
Used to run validation checks that require comparing multiple fields (e.g., checking if password and
confirm_password match).
Syntax:
@model_validator(mode="after")
def verify_password_match(self) -> "ModelClass":
if [Link] != self.confirm_password:
raise ValueError("Passwords do not match")
return self
5. Practical Examples
This example features a user registration payload with password strength checks and email
normalization.
import re
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, EmailStr, field_validator, model_validator
app = FastAPI()
class UserRegisterSchema(BaseModel):
# Field constraints
username: str = Field(..., min_length=3, max_length=20, pattern=r"^[a-zA-Z0-
9_]+$")
email: EmailStr # Enforces valid email formats
password: str = Field(..., min_length=8)
confirm_password: str = Field(..., min_length=8)
@[Link]("/register")
async def register(payload: UserRegisterSchema):
return {"status": "validated", "username": [Link], "email":
[Link]}
6. Production Best Practices
If your API stores text entered by users that will be rendered on a frontend, sanitize the input strings
(e.g., stripping HTML tags) using custom validators to prevent Cross-Site Scripting (XSS).
import html
from pydantic import field_validator
When raising custom validation errors inside field validators, never return sensitive data (like secret
tokens or parts of passwords) inside the raised ValueError message. These messages are returned
directly to the client.
7. Common Mistakes
# Antipattern: Redundant!
@[Link]("/items")
async def create_item(item: Item):
try:
# FastAPI already validates before this function runs!
pass
except ValidationError:
pass
Correction: Let FastAPI handle validation. If you need custom exception formats, override the global
RequestValidationError handler.
8. Performance Tips
class NetworkConfig(BaseModel):
ip_address: str
@field_validator("ip_address")
@classmethod
def validate_ip(cls, val: str) -> str:
if not IP_PATTERN.match(val):
raise ValueError("Invalid IP format")
return val
9. Security Considerations
When writing unit tests or debugging API errors, you can catch Pydantic's native exception structure and
print it in a human-readable format:
try:
UserRegisterSchema(username="ab", email="invalid", password="123",
confirm_password="123")
except ValidationError as e:
# Print formatted error logs
print([Link]())
Users often accidentally copy-paste leading or trailing spaces into input forms (e.g. " alice_username
" ). Stripping whitespace globally ensures database lookups succeed.
class ProfileUpdate(BaseModel):
bio: str = Field(..., max_length=500)
@field_validator("bio", mode="before")
@classmethod
def strip_spaces(cls, val: str) -> str:
if isinstance(val, str):
return [Link]()
return val
Answer:
@field_validator operates on a single field during the validation process. It is used to validate or
sanitize specific field properties (like checking password length or formatting email addresses).
@model_validator operates on the entire model data structure. It runs either before
( mode="before" ) or after ( mode="after" ) field-level validation and is used for cross-field verification
(such as validating that start_date is earlier than end_date ).
Question 2: How does Pydantic type coercion work, and how can you enforce strict
type matching?
Answer:
Pydantic attempts to coerce inputs to the declared target type if it can do so safely (e.g., converting a
string "42" to an int or "true" to a bool ).
To disable coercion and enforce strict type verification, set strict=True inside the model configuration,
or use the Strict type aliases:
class StrictUser(BaseModel):
# Only accepts native integers; strings like "42" will be rejected
age: StrictInt
13. Exercises
Create a schema DateRangeSchema with start_date and end_date attributes. Add a validator to
ensure end_date is strictly after start_date .
Solution:
class DateRangeSchema(BaseModel):
start_date: date
end_date: date
@model_validator(mode="after")
def verify_range(self) -> "DateRangeSchema":
if self.end_date <= self.start_date:
raise ValueError("end_date must be after start_date")
return self
This mini-project is a financial transaction record parser. It validates transaction directions ( credit /
debit ), enforces positive transaction limits, normalizes currency codes, and verifies that the transaction
amounts align with target payment modes.
class TransactionPayload(BaseModel):
transaction_id: str = Field(..., pattern=r"^TXN-[A-Z0-9]{8}$")
account_number: str = Field(..., pattern=r"^\d{10,12}$")
direction: Literal["credit", "debit"]
amount: float = Field(..., gt=0.0)
currency: str = Field(default="USD")
payment_mode: Literal["wire", "card", "atm"]
timestamp: datetime = Field(default_factory=[Link])
return self
# Valid Payload
valid_data = {
"transaction_id": "TXN-A1B2C3D4",
"account_number": "123456789012",
"direction": "debit",
"amount": 250.00,
"currency": "eur",
"payment_mode": "card"
}
parsed_txn = TransactionPayload(**valid_data)
print(f"Transaction verified: {parsed_txn.transaction_id}")
print(f"Currency coerced and normalized: {parsed_txn.currency}") # "EUR"
try:
TransactionPayload(**invalid_atm_data)
except ValueError as e:
print(f"Validation intercepted invalid ATM payload: {e}")
Key Takeaways
1. Input validation guards application boundaries, converting raw client payloads into strictly typed
Python structures.
2. Pydantic performs smart type coercion by default, converting parameters (like strings to integers)
when safe. Enforce strict type-matching using model configurations if needed.
3. Use @field_validator to validate and clean individual parameters, and @model_validator to run
checks across multiple fields.
4. Enforce size limits ( max_length / le ) on collections and numeric fields to protect your application
against Denial of Service (DoS) memory crashes.
Further Reading
1. Introduction
Pydantic is the data validation backbone of modern Python web frameworks. However, in Pydantic v1,
all validation and serialization logic was written in pure Python. While flexible, this introduced
performance bottlenecks, especially under high load or when parsing large JSON payloads.
To solve this, Pydantic v2 was completely rewritten. The core validation engine was extracted into a
separate project, pydantic-core , written in Rust. This architectural shift resulted in substantial
improvements:
For FastAPI developers, upgrading to Pydantic v2 (integrated in FastAPI since version 0.100.0) means
immediate, out-of-the-box performance gains for all endpoints.
2. Theory
Pydantic v2 acts as a thin Python wrapper around pydantic-core . When you declare a Pydantic model:
4. When a request arrives, the raw data is validated and parsed directly in Rust, returning the validated
Python objects.
In v1, converting an object to a dictionary was slow because it walked the object tree recursively in
Python. In v2, serialization is handled by the Rust engine, making methods like model_dump()
extremely fast.
3. Internal Working
Memory Representation
When Pydantic validates a model, it bypasses standard Python dictionary parsing where possible. By
calling model_validate_json() with a raw JSON string, Pydantic parses the string directly into Python
types in Rust, avoiding the overhead of creating intermediate Python dictionaries.
Raw JSON String ---> [Rust Parser (pydantic-core)] ---> Validated Python Model
Object
(Bypasses intermediate Python dictionaries creation!)
4. API Reference
Pydantic v2 deprecated or renamed several core methods to clean up the API surface.
Concept Legacy (v1) Modern (v2)
Instead of defining a nested class named Config , v2 uses a class attribute named model_config typed
with ConfigDict .
class User(BaseModel):
model_config = ConfigDict(
str_strip_whitespace=True, # Automatically strip whitespace
extra="forbid", # Reject fields not defined in the model
populate_by_name=True # Allow using field aliases during
initialization
)
username: str
5. Practical Examples
This example illustrates the core v2 APIs for validation, serialization, configurations, and schema
generation.
# 3. Serialization to Dictionary
print("Serialized Dict:", product.model_dump())
# Output: {'name': 'Mechanical Keyboard', 'price': 89.99, 'sku_code': 'KB-88'}
Set extra="forbid" to prevent clients from sending undocumented parameters, which could be an
attempt to inject data or pollute database tables.
Set str_strip_whitespace=True globally to sanitize string inputs before storing them in your
database.
class StrictModel(BaseModel):
model_config = ConfigDict(
extra="forbid",
str_strip_whitespace=True
)
7. Common Mistakes
8. Performance Tips
In custom middleware or websocket message receivers, you often receive raw JSON bytes or strings.
Avoid converting the string to a dictionary using [Link]() before validation. Instead, pass the raw
JSON directly to model_validate_json() :
Slow Path:
import json
parsed_dict = [Link](raw_payload) # Slow Python parser
model = UserModel.model_validate(parsed_dict)
9. Security Considerations
1. Limiting Validation Nesting Depth
If your models support deep nesting (e.g. recursive category trees), a client could send a payload
nested hundreds of levels deep. Running validation on this can cause a stack overflow in the interpreter.
Mitigation: Pydantic limits recursion depth by default. Avoid overrides that bypass these
boundaries.
When debugging validation issues, you can inspect the structured error list programmatically:
try:
ProductSchema(name="A", price=-5.0, skuCode="SKU")
except ValidationError as e:
# returns an array of structured error dictionaries
error_list = [Link]()
for err in error_list:
print(f"Location: {err['loc']} | Issue: {err['msg']} | Code:
{err['type']}")
When returning database rows query models (such as SQLAlchemy entities) from route endpoints,
configure from_attributes=True (which replaces orm_mode=True from Pydantic v1) to serialize
database objects easily.
class UserDB:
"""Mock database entity."""
def __init__(self, id: int, username: str):
[Link] = id
[Link] = username
class UserResponse(BaseModel):
model_config = ConfigDict(from_attributes=True) # Enables ORM loading
id: int
username: str
# Usage:
db_user = UserDB(id=1, username="bob")
# Read attributes dynamically from class object
response = UserResponse.model_validate(db_user)
Answer:
Pydantic v2's validation and serialization engine ( pydantic-core ) is written in Rust. Type hints are
compiled into a validation schema at startup. When validating inputs, the data is processed directly in
compiled Rust code rather than walking the object tree recursively in Python. This reduces CPU
overhead and garbage collection pressure, especially for nested models and large JSON strings.
Question 2: How do you handle field aliases in Pydantic v2 when validating inputs vs
serializing outputs?
Answer:
You define aliases using Field(..., alias="alias_name") .
By default, validation ( model_validate ) expects the alias key name (e.g., skuCode ).
If you want to allow initialization using both the attribute name ( sku_code ) and the alias, configure
populate_by_name=True in ConfigDict .
When serializing output, model_dump() uses Python names by default. To output using the alias
names, pass by_alias=True : model.model_dump(by_alias=True) .
13. Exercises
Create a model DatabaseConfig with host: str and port: int . Configure the model to strip
whitespace, forbid extra arguments, and run validation in strict mode (raising an error if the port is sent
as a string).
Solution:
class DatabaseConfig(BaseModel):
model_config = ConfigDict(
strict=True,
extra="forbid",
str_strip_whitespace=True
)
host: str
port: int
This mini-project is a high-performance batch validation pipeline. It processes large arrays of JSON
strings concurrently using Pydantic v2's Rust-backed JSON validator and compiles detailed success
and error reports.
class UserProfile(BaseModel):
model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
class BatchProcessor:
def __init__(self, raw_json_dataset: list[str]):
[Link] = raw_json_dataset
[Link]: list[UserProfile] = []
[Link]: list[dict] = []
return {
"processed_count": len([Link]),
"success_count": len([Link]),
"failure_count": len([Link]),
"elapsed_ms": elapsed * 1000
}
raw_payloads = [
'{"username": "alice", "email": "alice@[Link]", "age": 28}',
'{"username": "bob", "email": "invalid-email", "age": 32}', # Invalid email
'{"username": "ch", "email": "charlie@[Link]", "age": 17}', # Username
short, Under age
'{"username": "david", "email": "david@[Link]", "age": 45, "extra_key":
"spam"}' # Extra key
]
processor = BatchProcessor(raw_payloads)
summary = [Link]()
Key Takeaways
1. Pydantic v2 uses a validation engine written in Rust ( pydantic-core ), making data validation and
serialization significantly faster.
2. In v2, configurations are defined using model_config = ConfigDict(...) as class attributes rather
than nested class definitions.
3. For maximum parsing performance, pass raw JSON strings directly to model_validate_json() to
bypass intermediate Python dictionary creation.
4. Set from_attributes=True inside ConfigDict to serialize ORM objects (like database rows) into
Pydantic models.
Further Reading
1. Introduction
When a client (like a web browser or a mobile app) makes a request to a FastAPI application, a
sequence of operations is triggered. The request is parsed, routed, validated, processed by
dependencies, handled by your business logic, serialized, and returned as a response.
Understanding the Request Lifecycle—the exact sequence and timing of these operations—is crucial
for:
Debugging: Knowing where in the chain an error occurred (e.g., did it fail in middleware, validation,
or dependency injection?).
Performance Tuning: Understanding how many times a dependency runs and when resources are
opened/closed.
In this chapter, we will trace the journey of an HTTP request from the moment the socket receives TCP
bytes to the moment the response is sent back.
2. Theory
The request lifecycle can be thought of as a onion-like wrapper structure. The request starts from the
outside, passes through layers of middleware, reaches the inner core (your route handler), and then
flows back out through the same middleware layers.
[Client Request]
|
v
+---------------------------------------+
| Layer 1: ASGI Server (Uvicorn) | <-- Socket parsing, Scope creation
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 2: Middleware Stack | <-- CORS, Logging, Sessions (Request
path)
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 3: Routing Resolution | <-- URL path matching, 404/405 checks
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 4: Input Validation | <-- Query/Body parsing, Pydantic check
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 5: Dependency Graph Resolution | <-- Security, DB session injection
+---------------------------------------+
|
v
[ROUTE HANDLER FUNCTION] <-- Your Endpoint Logic executes
|
v
+---------------------------------------+
| Layer 6: Dependency Cleanup (yield) | <-- Database session closes
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 7: Response Serialization | <-- Pydantic converts objects to JSON
+---------------------------------------+
|
v
+---------------------------------------+
| Layer 8: Middleware Stack | <-- Custom headers, response logging
(Response path)
+---------------------------------------+
|
v
[Client Response]
3. Internal Working
The ASGI server (e.g., Uvicorn) receives TCP packets on the bound port. It parses the raw bytes into
HTTP protocol representations, constructs the ASGI scope dictionary, and registers receive / send
hooks.
The request enters the middleware stack. Middlewares process requests sequentially in the order they
were registered. A middleware can inspect headers, log incoming metadata, or block unauthorized
requests early.
Starlette's router inspects the scope["path"] and matches it against the compiled regex paths.
If the path matches but the HTTP method (e.g., POST instead of GET) is incorrect, it raises a 405
Method Not Allowed .
FastAPI extracts raw variables from path segments, query parameters, headers, cookies, and the JSON
request body. It runs validation using Pydantic schemas. If a field fails validation, a
RequestValidationError exception is raised, and execution skips the handler.
FastAPI resolves the dependency graph for the route. It executes dependencies sequentially, caching
shared dependencies (by default). If a dependency fails or raises an HTTPException , execution is
halted, and cleanup steps are called immediately.
Your route handler function is executed. If it was defined as async def , it runs directly on the event
loop. If defined as def , it is offloaded to the Starlette worker threadpool.
Step 7: Yield Dependency Cleanup
FastAPI resumes execution of any generator-based dependencies. The code following the yield
statement runs to clean up resources (e.g., closing database connections).
The values returned by your handler are serialized using Pydantic models. This converts Python objects
into JSON-compatible primitives (dicts/lists).
The response travels back through the middleware stack in reverse order. Middlewares can append
custom security headers (e.g. CORS), modify status codes, or compress the response payload before
passing it to the server.
Uvicorn translates the ASGI response events into raw HTTP bytes and writes them back to the client
socket.
4. API Reference
While the request lifecycle is handled automatically, FastAPI provides hooks to hook into it:
Request : The raw Starlette request object, exposing request properties ( headers , cookies ,
client , state ).
Response : The output class. You can inject custom response instances directly into endpoints.
5. Practical Examples
This example registers custom logs at different stages of the request execution to visualize the lifecycle
path.
import time
from fastapi import FastAPI, Depends, Request, Response
from typing import Annotated
app = FastAPI()
# 3. Endpoint
@[Link]("/items/{item_id}")
async def read_item(
item_id: int,
db: Annotated[str, Depends(get_db)]
):
print(f" [Lifecycle] 4. Route Handler executing for item: {item_id}")
return {"item_id": item_id, "source": db}
Because middlewares execute for every single request, any slow synchronous execution or un-
optimized DB call inside a middleware impacts your entire API. Keep middleware logic lean. For heavy
verification checks (like querying databases for roles), use FastAPI dependencies instead, which run
only on the routes that declare them.
7. Common Mistakes
Mistake: Assuming that code after yield in a dependency runs after the middleware returns the
response.
Correction: The generator cleanup block runs before the response leaves the middleware stack.
Middlewares execute around the entire routing and dependency loop. Do not attempt to use resources
yielded by dependencies inside the response path of your middlewares.
8. Performance Tips
If your route depends on both get_current_user and verify_premium_status , and both need the
database, ensure they share the same database dependency:
FastAPI's dependency caching resolves get_db once, saving the overhead of opening multiple
connection pools within the same request.
9. Security Considerations
FastAPI stops execution immediately if Pydantic validation fails. This is a critical security shield,
preventing invalid data from reaching your business logic or database. Ensure that all query and body
inputs have strict schemas. Relying on broad types like dict or Any bypasses this validation layer,
exposing your code to injection or runtime crashes.
If you need to trace request-level metadata (such as an execution ID or correlation token) across
logging boundaries, attach it to [Link] . The [Link] object persists across the entire
request scope.
@[Link]("http")
async def add_correlation_id(request: Request, call_next):
# Attach tracking ID to request state
[Link].correlation_id = "TXN-999"
response = await call_next(request)
return response
@[Link]("/data")
async def get_data(request: Request):
# Retrieve tracking ID inside route handler
cid = getattr([Link], "correlation_id", None)
return {"correlation_id": cid}
In distributed systems, microservices must pass a correlation ID through HTTP headers to trace
requests across services. By extracting this ID at the middleware layer, storing it in [Link] , and
referencing it in logging utilities, you get a clean, end-to-end trace of your application.
Question 1: Describe the exact sequence of events when a client request hits a FastAPI
application.
Answer:
1. The ASGI server receives raw TCP packets and parses them into HTTP representation.
2. The request passes through the HTTP middleware stack.
9. The response passes back through the middleware stack in reverse order.
10. The ASGI server formats the response as HTTP bytes and writes them to the socket.
Question 2: Why do dependencies run after middlewares? Under what scenario does a
middleware run but dependencies do not?
Answer:
Middlewares sit at the outer edge of the request onion structure, handling raw HTTP packets before the
router runs. Dependencies are defined at the route level, meaning FastAPI must match a route before it
can resolve its dependency graph.
If the client requests a URL that does not exist (resulting in a 404 Not Found ) or sends a request
matching a route with an unsupported HTTP method (resulting in a 405 Method Not Allowed ), the
middleware runs, but route resolution fails, meaning the dependency injection graph is never executed.
13. Exercises
Write a middleware that logs: [INFO] GET /items/5 Started at the beginning of a request, and [INFO]
GET /items/5 Finished - Status 200 upon completion.
Solution:
@[Link]("http")
async def trace_requests(request: Request, call_next):
method = [Link]
path = [Link]
print(f"[INFO] {method} {path} Started")
This mini-project implements a complete tracing utility. It tracks requests, measures processing times,
and records logs at different stages of the request lifecycle.
import uuid
import time
from fastapi import FastAPI, Depends, Request, Response
from typing import Annotated
app = FastAPI()
[Link]["X-Trace-Id"] = trace_id
return response
# 4. Target Endpoint
@[Link]("/items")
async def list_items(tracker: Annotated[RequestTracker, Depends(get_tracker)]):
[Link]("Route Handler executing")
return {"status": "ok", "items": [1, 2, 3]}
Key Takeaways
1. The request lifecycle follows an onion-like structure, starting from the ASGI server, moving through
middlewares, the router, validation, and dependencies, and then returning in reverse order.
2. Middlewares execute for all incoming requests, making them ideal for cross-cutting tasks but
sensitive to performance overhead.
3. Path validation and dependency graph resolution only occur after a route is successfully matched.
4. Use [Link] to pass metadata (like correlation IDs) across the request lifecycle boundaries.
Further Reading
1. Introduction
In a web application, there are operations that must run for every incoming request, regardless of which
endpoint is targeted. Examples include:
Security: Injecting HTTP headers to protect against cross-site scripting (XSS) or clickjacking.
CORS: Enforcing Cross-Origin Resource Sharing rules to control which frontend domains can query
your API.
Observability: Measuring request latency and logging transaction metrics.
Middleware is a layer that wraps your application. It intercepts incoming requests, performs pre-
processing tasks, passes the requests down the pipeline, intercepts the outgoing responses, and
performs post-processing tasks before returning them to the client.
In this chapter, we will explore FastAPI's middleware layers, learn how to configure built-in middlewares,
and build custom middlewares using both high-level decorators and low-level ASGI specifications.
2. Theory
How it works: A class implementing the standard ASGI protocol __call__(scope, receive,
send) method.
Pros: Extremely fast, low-overhead, and supports all protocols (HTTP, WebSockets, Lifespan
events).
Cons: Requires working directly with raw ASGI events and byte streams, which is more
complex.
3. Internal Working
Middlewares are evaluated as a stack of nested wrappers. When a request arrives, it enters the
outermost middleware, moves inward through the stack, reaches the router, and then bubbles back out
in reverse order.
Incoming Request ---> [Gzip Middleware] ---> [CORS Middleware] ---> [FastAPI
Router]
|
Endpoint Logic
|
Outgoing Response <--- [Gzip (Compressed)] <--- [CORS (Headers)] <--- [JSON Result]
If a middleware intercepts the request early and returns a response (e.g., CORS rejecting an
unauthorized cross-origin request), execution halts, and the request never reaches the router or
endpoints.
4. API Reference
Enforces Cross-Origin Resource Sharing policies to protect your API from unauthorized cross-origin
browser requests.
Configuration:
app.add_middleware(
CORSMiddleware,
allow_origins=["[Link]
allow_credentials=True,
allow_methods=["GET", "POST"],
allow_headers=["*"]
)
2. TrustedHostMiddleware
Enforces that all incoming requests contain a valid host header matching a whitelist. This protects your
API against HTTP Host Header attacks.
Configuration:
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["[Link]", "*.[Link]"]
)
3. GZipMiddleware
Compresses response payloads using Gzip if the client sends an Accept-Encoding: gzip header and
the response size exceeds a minimum threshold.
Configuration:
app.add_middleware(GZipMiddleware, minimum_size=1000)
5. Practical Examples
This example injects standard security headers into every outgoing response to harden the API.
app = FastAPI()
@[Link]("http")
async def inject_security_headers_middleware(request: Request, call_next):
# 1. Let the request execute down the pipeline
response = await call_next(request)
return response
@[Link]("/")
async def index():
return {"status": "secure"}
ALLOWED_ORIGINS = [
"[Link]
"[Link]
]
In FastAPI, middlewares are executed in the reverse order of registration (i.e. the last one added using
add_middleware is the first to process incoming requests). Ensure CORSMiddleware is added last so it
can catch and respond to CORS preflight options requests before other processing middlewares
intercept them.
7. Common Mistakes
Mistake: Modifying or reading the response body inside a custom @[Link]("http") function.
Correction: Avoid reading response bodies inside middlewares. If you must inspect payloads, use raw
ASGI middleware or handle logging at the controller/route level.
If you register Gzip compression before CORS, the response headers might be compressed and
packaged before CORS can inject origin validation details. This can lead to browsers blocking the
request due to missing CORS headers.
Correction: Always register CORS last (making it execute first on the request path and last on the
response path).
8. Performance Tips
9. Security Considerations
If an endpoint does not require authentication (e.g. a public blog article feed), setting
allow_origins=["*"] is safe.
If the endpoint processes credentials or session states, specify exact origin protocols ( https:// )
and domains. Avoid dynamic mirroring patterns (e.g., reading the request Origin header and
echoing it back in Access-Control-Allow-Origin ), as this bypasses origin protections.
If a middleware raises an exception before call_next is reached, FastAPI's router cannot execute. The
server returns a generic 500 Internal Server Error response, which can be hard to debug.
@[Link]("http")
async def safe_middleware(request: Request, call_next):
try:
return await call_next(request)
except Exception as e:
[Link](f"Middleware crashed: {e}", exc_info=True)
raise
A common real-world use case for raw ASGI middleware is performing sliding-window rate-limiting
checks (e.g., using Redis) at the edge of the application, rejecting requests before they run any
framework routing or parsing code.
Answer:
BaseHTTPMiddleware executes request handlers in a separate async task context to bridge Starlette's
response iterator. Because Python's contextvars module isolates state variables by task context, any
context variable set inside a BaseHTTPMiddleware (like setting a tenant ID or database session) is lost
when control returns to the main execution loop.
To preserve context variables across the request lifecycle, you should write a raw ASGI middleware
instead of using BaseHTTPMiddleware .
Question 2: In what order do middlewares execute on the request and response paths?
Answer:
Middlewares are executed as a nested stack:
Request Path: Executed in reverse order of registration (the last middleware added using
app.add_middleware runs first).
Response Path: Executed in order of registration (the first middleware registered is the last to
process the outgoing response).
13. Exercises
Write a middleware that generates a unique UUID request ID for every incoming request. Inject this ID
as a response header named X-Request-ID and attach it to the request scope so endpoints can access
it.
Solution:
import uuid
from fastapi import FastAPI, Request
app = FastAPI()
@[Link]("http")
async def inject_request_id(request: Request, call_next):
req_id = str(uuid.uuid4())
# Attach to request state context
[Link].request_id = req_id
14. Mini Project: Raw ASGI Security and Rate Limit Guard Middleware
This mini-project implements a high-performance, raw ASGI middleware that injects security headers,
logs request paths, and enforces rate-limiting boundaries, running independently of the high-level
FastAPI classes.
import time
from typing import Callable, Awaitable
class RawASGIGuardMiddleware:
def __init__(
self,
app,
rate_limit_per_minute: int = 120
):
[Link] = app
self.rate_limit = rate_limit_per_minute
# Simple in-memory tracker (In production, use Redis)
self.client_hits: dict[str, list[float]] = {}
# 2. Extract Client IP
client_host, _ = [Link]("client", (None, None))
if client_host:
# Enforce sliding window rate limit
now = [Link]()
hits = self.client_hits.setdefault(client_host, [])
# Prune hits older than 60 seconds
hits[:] = [h for h in hits if now - h < 60.0]
[Link](now)
event["headers"] = headers
await send(event)
# 4. Pass down processing using our secure send interceptor
await [Link](scope, receive, secure_send)
Key Takeaways
1. Middlewares allow running cross-cutting code (like security headers or logging) globally before
requests reach route endpoints.
2. CORSMiddleware protects APIs from unauthorized cross-origin requests. Whitelist specific domains
in production instead of using allow_origins=["*"] .
4. Raw ASGI middleware operates directly on lower-level ASGI events, offering high performance and
compatibility with all protocols (including WebSockets).
Further Reading
1. Introduction
In a production application, errors are inevitable. Databases go down, clients send incorrect resource
IDs, payment gateways timeout, or file uploads exceed storage limits.
1. Bad User Experience: The client receives a generic crash response (like a blank page or a raw
stack trace) instead of a helpful error message.
2. Security Vulnerabilities: Raw traceback details can expose database structures, internal paths,
package versions, and credentials to attackers.
3. Operational Blindness: If exceptions are not intercepted and logged correctly, developers cannot
monitor or fix issues.
Exception Handling is the practice of defining error boundaries. When an error occurs, the application
intercepts it, performs clean-up actions, logs the error, and returns a clean, structured JSON response
to the client.
FastAPI provides a global exception mapping system that lets you handle errors centrally, separating
error-handling logic from your route controllers.
2. Theory
Exception Boundaries
An Exception Boundary is a wrapper that catches any error bubble up from your application code. In a
clean architecture, you want database or domain errors to be caught before they escape, translating
them into HTTP status codes at the API layer.
[API Border] <--- Intercepts and formats JSON error <--- (Exception bubbles up)
|
[Endpoints Route]
|
[Service Layer]
|
[Database Layer] (Raises EntityNotFoundError)
By defining global boundaries, you ensure that even if a developer forgets to write a try/except block
inside an endpoint, the application will not expose sensitive crash details.
3. Internal Working
Under the hood, FastAPI wraps the application routing table inside Starlette's ExceptionMiddleware .
When an exception occurs during request execution:
1. The exception propagates up the call stack, exiting your endpoint and dependencies.
3. The middleware checks if a handler is registered for that specific exception class or any of its parent
classes.
5. If no handler matches, the exception is re-raised, escaping to the ASGI server layer, which logs the
traceback and returns a generic 500 Internal Server Error response.
4. API Reference
Used to register an async function as the handler for a specific exception class.
Syntax:
@app.exception_handler(MyException)
async def my_exception_handler(request: Request, exc: MyException) -> Response:
return JSONResponse(status_code=400, content={"detail": [Link]})
FastAPI's HTTPException inherits from Starlette's version but adds a headers parameter. This is
critical for authentication endpoints where you must return headers like WWW-Authenticate .
Rule: Always import HTTPException from fastapi inside your endpoint files:
from fastapi import HTTPException .
5. Practical Examples
app = FastAPI()
class InsufficientFundsError(Exception):
def __init__(self, required: float, available: float):
[Link] = required
[Link] = available
@app.exception_handler(InsufficientFundsError)
async def insufficient_funds_handler(request: Request, exc:
InsufficientFundsError):
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={
"error_code": "INSUFFICIENT_FUNDS",
"message": "Transaction declined due to insufficient account balance.",
"details": {"required": [Link], "available": [Link]}
}
)
# 3. Route Endpoint
@[Link]("/users/{username}/withdraw")
async def withdraw(username: str, amount: float):
# Simulate DB search
if username != "alice":
raise UserNotFoundError(username)
# Simulate balance check
balance = 100.0
if amount > balance:
raise InsufficientFundsError(required=amount, available=balance)
In production, ensure that global exception handlers do not return the traceback parameter
( traceback.format_exc() ) to the client. Instead:
1. Log the full traceback internally using a logging utility to help developers debug.
2. Return a clean, generic message to the client, such as "Internal Server Error" , along with a
unique reference ID.
import logging
import uuid
logger = [Link]("app")
@app.exception_handler(Exception)
async def global_fallback_handler(request: Request, exc: Exception):
error_id = str(uuid.uuid4())
# Log the full traceback internally
[Link](f"Unhandled error [{error_id}]: {exc}", exc_info=True)
# Return generic error response to the client
return JSONResponse(
status_code=500,
content={
"error_code": "INTERNAL_SERVER_ERROR",
"message": "An unexpected error occurred. Please contact support.",
"reference_id": error_id
}
)
7. Common Mistakes
Mistake: Catching a broad exception class and returning a 200 OK status code.
If a handler catches a database connection failure but returns 200 OK with an error message in the
response body, caching proxies or API gateways might treat it as a successful response, caching the
error payload for other users.
Correction: Always return appropriate HTTP status codes (e.g., 5xx for server crashes, 4xx for client
inputs) to indicate the true status of the request.
8. Performance Tips
Exceptions in Python are relatively expensive to raise and catch due to the overhead of creating and
unwinding the stack trace.
Do not raise an exception to handle expected flow control (like checking if a value is present in a
list). Use standard conditional statements ( if/else ) instead.
Reserve exceptions for exceptional situations (e.g. database disconnects or security failures).
9. Security Considerations
If your database driver raises a validation constraint error (like a PostgreSQL foreign key violation), do
not return the raw database error message to the client.
An error message like "insert or update on table 'orders' violates foreign key constraint
'fk_customer_id'" exposes internal database table structures and relations.
Mitigation: Catch database errors at the service layer and translate them into generic application
exceptions (like InvalidOrderRequest ) before they reach the API boundary.
When client inputs fail FastAPI validation, it logs a basic debug entry. In development environments, it is
helpful to customize the RequestValidationError handler to log incoming invalid payloads to help
frontend developers debug requests:
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc:
RequestValidationError):
# Log details locally
print(f"Validation failed for URL: {[Link]}")
print(f"Invalid Body: {[Link]}")
# Return default 422 JSON response
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content={"detail": [Link]()}
)
Large corporate APIs often require a standardized error response structure across all endpoints. Global
exception handlers allow you to format all HTTP, database, and validation errors into a single, unified
JSON structure.
Answer:
FastAPI's HTTPException inherits from Starlette's version but adds support for response headers via a
headers parameter. This is critical for security protocols (such as OAuth2 authentication, where the
server must return a WWW-Authenticate header in the response when credentials are missing or
invalid). Starlette's version lacks this parameter, which can make it harder to comply with security
specifications.
Question 2: How can you capture and customize validation error responses globally in
FastAPI?
Answer:
FastAPI validation errors raise a RequestValidationError exception. You can customize the response
by registering a custom exception handler for RequestValidationError :
@app.exception_handler(RequestValidationError)
async def custom_handler(request: Request, exc: RequestValidationError):
# Format and return custom JSON structure
return JSONResponse(status_code=422, content={"errors": [Link]()})
13. Exercises
Write a custom exception DatabaseConnectionError . Create a global exception handler that logs "DB
Down!" to the console and returns a 503 Service Unavailable response to the client.
Solution:
class DatabaseConnectionError(Exception):
pass
app = FastAPI()
@app.exception_handler(DatabaseConnectionError)
async def db_connection_error_handler(request: Request, exc:
DatabaseConnectionError):
print("[CRITICAL ERROR]: DB Down!")
return JSONResponse(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
content={"detail": "Database is temporarily unavailable. Please try again
later."}
)
This mini-project simulates a payment gateway client. It defines payment exceptions (insufficient funds,
payment timeouts, card validation issues), wraps client calls inside an error boundary, and maps
exceptions to structured HTTP responses.
# 1. Custom Exceptions
class PaymentFailure(Exception):
"""Base payment exception."""
pass
class CardExpiredError(PaymentFailure):
pass
class ProcessorTimeoutError(PaymentFailure):
pass
@app.exception_handler(ProcessorTimeoutError)
async def processor_timeout_handler(request: Request, exc: ProcessorTimeoutError):
# Return 504 Gateway Timeout
return JSONResponse(
status_code=status.HTTP_504_GATEWAY_TIMEOUT,
content={
"status": "failed",
"error_code": "PROCESSOR_TIMEOUT",
"message": "The payment processor timed out. Please try again."
}
)
# Instantiate Service
processor = PaymentProcessor()
# 5. Route Endpoint
@[Link]("/checkout")
async def checkout(card_number: str, amount: float):
# Execute transaction inside safety boundary
try:
charge_id = processor.charge_card(card_number, amount)
return {"status": "success", "charge_id": charge_id}
except PaymentFailure:
# Re-raise so registered global exception handlers can catch it
raise
except Exception as e:
# Fallback security handler
raise HTTPException(status_code=500, detail="An unhandled error occurred.")
Key Takeaways
1. Global exception boundaries catch errors at the API border, preventing raw tracebacks from leaking
database schemas, version numbers, or path structures.
2. FastAPI translates uncaught exceptions into standard HTTP responses using Starlette's
ExceptionMiddleware dispatcher.
3. FastAPI's HTTPException adds support for headers, making it the preferred choice over Starlette's
version for endpoints requiring authentication headers.
4. Catch low-level database errors at the service layer and translate them into generic application
exceptions to protect your API from schema leakage.
Further Reading
1. Introduction
When a client makes a request to your API, the primary goal is to return a response as quickly as
possible. However, some requests trigger operations that take a long time to complete:
If you perform these tasks directly inside your endpoint function, the client is forced to wait for them to
finish. For example, if sending an email takes 2.5 seconds, your API response is delayed by 2.5
seconds.
Background Tasks allow you to offload these slow tasks. You register the task, immediately return a
success response (like 202 Accepted ) to the client, and the application executes the task in the
background after the response has been sent.
In this chapter, we will learn how to use FastAPI's built-in BackgroundTasks utility, explore its threading
model, and compare it with distributed queue managers like Celery.
2. Theory
A background task is a function that is scheduled to run after the HTTP response is written to the client.
This means the client does not wait for the task to complete, freeing up the client connection.
It is important to understand the differences and trade-offs of FastAPI's built-in tasks vs. heavy task
queues:
Requires a broker
No extra services needed (runs in-
Infrastructure (Redis/RabbitMQ) and worker
process).
processes.
Resource Shares CPU and memory with the Workers run in separate processes
Isolation main web application. or servers, isolating resource load.
3. Internal Working
Endpoint schedules Task ---> Route returns Response ---> Server writes response to
socket
|
v
Event Loop runs Task in
background
1. You declare a parameter of type BackgroundTasks in your endpoint and call
background_tasks.add_task(task_func, *args) .
2. FastAPI returns the response payload and closes the client connection.
3. Once the response is sent, the event loop runs the registered task.
If the task is an async def function, it is executed directly on the event loop.
If the task is a synchronous def function, it is run inside Starlette's threadpool so it does not
block the main event loop thread.
4. API Reference
*args / **kwargs : The arguments and keyword arguments to pass to the function.
5. Practical Examples
This example simulates registering a user, immediately returning a success response, and queuing a
slow email-sending task in the background.
import asyncio
from fastapi import FastAPI, BackgroundTasks
app = FastAPI()
# 2. Route Endpoint
@[Link]("/register")
async def register_user(
username: str,
email: str,
background_tasks: BackgroundTasks
):
# Schedule the task to run in the background
background_tasks.add_task(send_welcome_email, email, username)
Since built-in background tasks run within the same application process, running CPU-heavy operations
(like image resizing or machine learning inference) will consume CPU resources on your web server.
This degrades the performance of your web API, increasing request times for all users.
Best Practice: Use built-in BackgroundTasks only for light I/O operations (like calling external APIs,
writing small log files, or sending emails). For CPU-heavy tasks, offload them to a distributed task
queue like Celery.
7. Common Mistakes
Mistake: Registering a synchronous task that performs blocking operations without declaring it as a
standard def function.
If you declare your background function as async def but perform blocking synchronous operations
inside it (like calling a legacy database driver or using [Link] ), it will run on the main event loop
thread, freezing the entire application.
Correction: Declare blocking background tasks as synchronous def functions so FastAPI executes
them in the worker threadpool.
Mistake: Relying on built-in BackgroundTasks for mission-critical operations (like financial processing).
If the server process crashes, gets killed, or restarts during a deployment while background tasks are
queued, those tasks are lost forever.
Correction: For critical operations that must complete, store the tasks in a database first, or use a
persistent task broker like Celery.
8. Performance Tips
If you queue a large number of synchronous background tasks, Starlette's threadpool can run out of
worker threads, causing subsequent requests to queue up.
Monitor your task execution rates, and ensure that if tasks are generated faster than they can be
executed, you migrate them to an external task queue to keep your web server responsive.
9. Security Considerations
When you register a background task, you pass parameters to it (e.g. user IDs or file paths). If these
parameters are derived from client inputs, ensure they are validated before being queued.
For example, if you schedule a background file deletion task, make sure the target path has been
sanitized to prevent a client from scheduling the deletion of critical system files.
Because background tasks execute after the response has been sent to the client, if a task crashes or
raises an exception, the client does not see it. The exception is logged to the console, but it does not
trigger your global HTTP exception handlers.
Remedy: Wrap background task logic in try/except blocks to log errors to your logging or error-
tracking systems (like Sentry):
import logging
logger = [Link]("app")
In banking or security applications, every critical operation (like transferring money or updating
passwords) must be recorded in an audit log database. Doing this in a background task ensures that
audit logging overhead does not slow down user operations.
Answer:
FastAPI's BackgroundTasks class runs in-process. Tasks are stored in memory and executed by
the same Python process running the event loop. If the process restarts, pending tasks are lost.
Celery runs out-of-process. Tasks are serialized, sent to an external broker (like Redis or
RabbitMQ), and executed by separate worker processes. It is persistent, supports automatic retries,
isolates resource consumption, and is designed for heavy, mission-critical tasks.
Answer:
If the task function is defined as async def , it executes directly on the main event loop thread. It
must be non-blocking to prevent freezing the server.
If the task function is defined as def , FastAPI runs it in Starlette's background threadpool, ensuring
it does not block the main event loop.
13. Exercises
Write an endpoint /log that accepts a query parameter message and schedules a background task to
write that message to a file named [Link] along with a timestamp.
Solution:
app = FastAPI()
@[Link]("/log")
async def log_message(message: str, background_tasks: BackgroundTasks):
background_tasks.add_task(write_log, message)
return {"status": "queued"}
14. Mini Project: Asynchronous File Archiver Service
This mini-project simulates a file backup utility. It receives list records, returns an immediate success
code, and zips the files in the background using threadpool delegation to prevent blocking the API.
import os
import zipfile
import time
from fastapi import FastAPI, BackgroundTasks, HTTPException
from pathlib import Path
app = FastAPI()
class ZipArchiverService:
def create_zip_archive(self, archive_name: str, target_files: list[str]):
"""Synchronous CPU/IO bound archiving method."""
print(f"[Archiver] Starting zip generation for {archive_name}.zip...")
[Link](3.0) # Simulate heavy file compression overhead
# Instantiate service
archiver = ZipArchiverService()
@[Link]("/archive")
async def archive_files(
archive_name: str,
background_tasks: BackgroundTasks
):
# Mock files to archive
files_to_compress = ["book_structure.md", "[Link]"]
return {
"status": "archiving_started",
"archive_file": f"{archive_name}.zip",
"message": "Compression is executing in the background."
}
Key Takeaways
1. Background Tasks allow offloading slow operations, letting endpoints return responses to clients
immediately without waiting.
2. FastAPI's built-in BackgroundTasks executes in-process using memory arrays, making it lightweight
but sensitive to process crashes.
3. Async tasks run directly on the event loop, while sync tasks run in Starlette's threadpool to prevent
loop starvation.
4. Offload resource-intensive or CPU-bound tasks to Celery or distributed queues to isolate web server
resources.
Further Reading
1. Introduction
One of the most popular features of FastAPI is its out-of-the-box generation of interactive API
documentation. When you write a FastAPI application, you do not need to write separate documentation
files. FastAPI uses the OpenAPI Specification (formerly known as Swagger) to document your
endpoints automatically.
/docs : You get Swagger UI, an interactive interface where developers can view endpoints and
execute real API requests directly from the browser.
/redoc : You get ReDoc, a clean, three-panel documentation portal designed for production-ready
consumer APIs.
This documentation is not generated by parsing comments. It is compiled dynamically from your Python
type hints and Pydantic validation schemas. This ensures your documentation is always in sync with
your actual code.
2. Theory
The OpenAPI Specification (OAS) is a standardized schema description format for REST APIs. It
defines a JSON or YAML structure that describes:
Because this schema is standardized, it can be parsed by third-party tools to automatically generate
client libraries (SDKs) in dozens of programming languages (e.g. using OpenAPI Generator).
3. Internal Working
When your FastAPI application boots up, it compiles the routing tree. When a client requests the
/[Link] path:
2. For each route, it extracts the HTTP method, path, tags, and summary.
3. It inspects the function parameters and extracts type hints.
4. For request body parameters and response objects, it calls Pydantic's JSON schema generator
( model_json_schema() ) to generate the corresponding JSON Schema representations.
5. It compiles these properties into a single dictionary following the OpenAPI spec format.
6. It caches this dictionary in memory ( app.openapi_schema ) so subsequent requests to
/[Link] are served instantly without recompilation.
4. API Reference
You can customize the metadata of your API by passing parameters to the FastAPI class constructor:
app = FastAPI(
title="Core Payment API",
description="Corporate payment processing API for transactional records.",
version="1.4.2",
terms_of_service="[Link]
contact={
"name": "API Engineering Support",
"email": "support@[Link]",
},
license_info={
"name": "Apache 2.0",
"url": "[Link]
}
)
@[Link](
"/payments",
summary="Charge a payment card",
description="Charges a validated credit card. **Warning**: Perform check for
double-charges before calling.",
response_description="Confirmation payload of the completed charge",
tags=["Transactions"]
)
async def process_payment():
...
5. Practical Examples
This example shows how to configure Pydantic schemas so that Swagger UI displays realistic mock
inputs and response examples.
When you open /docs , the input form for this model will be pre-populated with the example values,
making it easy for frontend developers to test.
If your API documentation is private but needs to be accessible to external partners or QA engineers,
protect the /docs and /redoc paths using a simple HTTP Basic Auth dependency:
@[Link]("/docs", include_in_schema=False)
async def get_swagger_documentation(username: str = Depends(authenticate_docs)):
return get_swagger_ui_html(openapi_url="/[Link]", title="Secure docs")
7. Common Mistakes
8. Performance Tips
If your routes contain endpoints returning complex structures (such as raw database schemas or
analytical matrices) that are not meant for client SDK consumption, set include_in_schema=False
inside the route decorator:
@[Link]("/metrics/raw-dump", include_in_schema=False)
async def raw_metrics():
# Runs, but is hidden from OpenAPI specs and Swagger UI
return {}
This reduces the size of the generated [Link] file, speeding up loading times for Swagger UI.
9. Security Considerations
When defining model properties, verify that password fields or secret token variables do not contain
default values in the schemas (e.g. password: str = "default_pass" ). Pydantic includes default
values in the JSON schema metadata, exposing these values to anyone who queries your OpenAPI
definition.
You can write a simple test script to verify that the OpenAPI schema compiles without errors:
def test_openapi_schema_generation():
client = TestClient(app)
response = [Link]("/[Link]")
assert response.status_code == 200
schema = [Link]()
assert "openapi" in schema
assert "paths" in schema
By exposing a clean, standardized [Link] file, you can integrate tools like openapi-generator-
cli into your CI/CD pipeline. Every time the backend code updates, the pipeline automatically compiles
and publishes updated TypeScript, Dart, or Swift client libraries for your frontend and mobile teams.
Question 1: How does FastAPI compile Pydantic models into the OpenAPI schema?
Answer:
FastAPI leverages Pydantic's built-in JSON Schema generation. At startup, FastAPI inspects the
parameter annotations of your route handlers. If a parameter is a Pydantic model class, it calls
Model.model_json_schema() to get its JSON Schema representation. FastAPI then maps this
representation to the components/schemas section of the OpenAPI schema dictionary.
Answer:
Set include_in_schema=False in the route decorator:
@[Link]("/internal-status", include_in_schema=False)
async def check_internal_status():
...
The endpoint remains accessible to HTTP requests, but it is excluded from the /[Link] file,
hiding it from Swagger UI and ReDoc.
13. Exercises
Override the default [Link] method to inject a custom license footer {"x-security-tier":
"Enterprise"} into the OpenAPI schema metadata.
Solution:
app = FastAPI()
def custom_openapi():
# Implement caching check
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="Custom API",
version="1.0.0",
routes=[Link],
)
app.openapi_schema = openapi_schema
return app.openapi_schema
[Link] = custom_openapi
This mini-project overrides the default OpenAPI generator. It dynamically injects a global bearer token
security scheme (JWT) into all endpoints, inserts generic error components, and configures custom
branding.
from fastapi import FastAPI
from [Link] import get_openapi
from pydantic import BaseModel
class MoneyTransfer(BaseModel):
recipient_account: str
amount: float
@[Link]("/transfer", tags=["Banking"])
async def transfer_money(transfer: MoneyTransfer):
return {"status": "success"}
def custom_openapi_schema():
# Cache lookup
if app.openapi_schema:
return app.openapi_schema
Key Takeaways
1. FastAPI automatically compiles interactive API documentation (Swagger UI and ReDoc) from type
hints and Pydantic schemas.
2. The generated schema follows the standardized OpenAPI Specification, enabling auto-generation of
client SDKs.
3. Configure application-wide metadata using the FastAPI constructor, and route-level documentation
using parameters in route decorators.
4. Protect your API docs in production using Basic Authentication or by setting docs_url=None to hide
documentation paths.
Further Reading
1. Introduction
When you start learning FastAPI, it is common to write your entire application in a single file, usually
named [Link] . This is fine for small prototypes. However, as your application grows, a single-file
structure becomes unmaintainable:
Multiple developers working on the same file face constant code merge conflicts.
Circular import errors occur when routing, database initialization, and logic models depend on each
other.
To scale an application, you must apply the Separation of Concerns (SoC) principle. This involves
dividing your code into distinct layers (directories and files), where each module has a single, well-
defined responsibility.
In this chapter, we will design a production-ready project structure for FastAPI, learn how to manage
environments, and build a modular application skeleton.
2. Theory
Responsible for HTTP concerns: path registration, request payload validation (Pydantic), and
formatting responses. It should not run raw database queries or core business logic directly.
Implements the core business rules of your application (e.g., calculating pricing, registering
users, calling external services). This layer is independent of HTTP or web framework
specifications.
Handles database connections, ORM mapping structures (like SQLAlchemy entities), and
migration scripts.
3. Internal Working
Python Packaging and imports
Python packages are directories containing modules (files) and an __init__.py file (which can be
empty). When structuring a modular project:
Absolute Imports: Always reference imports from the root of the project (e.g. from
[Link] import settings ). This is the cleanest approach and prevents path resolution
errors when running tests or workers.
Avoid Circular Imports: Do not import the main app = FastAPI() instance inside your sub-routers.
Instead, define sub-routers using APIRouter() independently, and import and mount them to the
main app instance inside the central assembly file.
4. API Reference
Managing configuration variables (like database credentials, API secrets, and debug flags) is best
handled using Pydantic Settings. It reads configuration keys from environment variables or a .env file,
validates their types, and caches them.
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
5. Practical Examples
my_project/
├── .env # Local environment configuration file
├── .gitignore # Git exclusion mapping file
├── [Link] # Package dependencies and tool settings
├── app/
│ ├── __init__.py
│ ├── [Link] # Application assembly and startup
│ ├── api/ # API Web Routing layer
│ │ ├── __init__.py
│ │ ├── v1/
│ │ │ ├── __init__.py
│ │ │ ├── [Link] # Combines sub-routers
│ │ │ └── endpoints/
│ │ │ ├── __init__.py
│ │ │ ├── [Link]
│ │ │ └── [Link]
│ ├── core/ # Configuration and utilities
│ │ ├── __init__.py
│ │ ├── [Link] # Pydantic settings configuration
│ │ └── [Link] # Database session providers
│ ├── models/ # Database ORM models (SQLAlchemy)
│ │ ├── __init__.py
│ │ └── db_models.py
│ ├── schemas/ # Input/Output validation models (Pydantic)
│ │ ├── __init__.py
│ │ └── user_schemas.py
│ └── services/ # Business logic layer
│ ├── __init__.py
│ └── user_service.py
Never commit environment files ( .env ) or passwords directly to Git repositories. Add .env to your
.gitignore file.
In production, load configurations directly from the environment variables of your hosting platform
(Docker, AWS, Kubernetes, etc.), rather than reading a local .env file.
Configure your execution environment so the root folder is added to Python's system path. Run your
server using:
This sets the execution root directory correctly, ensuring that absolute imports like from
[Link] import settings resolve without throwing ModuleNotFoundError crashes.
7. Common Mistakes
Mistake: Declaring app = FastAPI() in [Link] , and importing app inside routers/[Link] to
register route decorators: @[Link]("/users") .
This causes a circular import because [Link] must import the router to mount it, but the router
imports [Link] to reference app .
Correction: Always define routers using APIRouter() inside sub-router modules:
# In app/api/v1/endpoints/[Link]
from fastapi import APIRouter
router = APIRouter()
@[Link]("/")
async def list_users(): ...
8. Performance Tips
Instantiating your Settings class reads file systems or checks environment structures, which can be
slow if done on every request.
Best Practice: Instanciate settings once globally inside your config module, or declare it as a
cached dependency using @lru_cache :
@lru_cache
def get_settings() -> Settings:
return Settings()
9. Security Considerations
Ensure your .gitignore file includes patterns to block sensitive files from being pushed to public
version control:
# gitignore
.env
.venv/
__pycache__/
*.pyc
*.db
This crash occurs if you run Python from inside the app/ subfolder instead of the root directory.
Remedy: Always run your commands from the root directory (where app/ is a subdirectory) using
the module execution syntax:
python -m uvicorn [Link]:app
Alternative: Add your project root folder to the PYTHONPATH environment variable in your terminal:
set PYTHONPATH=. (Windows) or export PYTHONPATH=. (Linux/macOS).
Multi-Environment configuration
Real-world systems deploy code across multiple environments (Development, Staging, Production). You
can structure Pydantic Settings to automatically load the correct configurations based on an
environment flag:
import os
from pydantic_settings import BaseSettings, SettingsConfigDict
class BaseConfig(BaseSettings):
db_name: str = "dev_db"
class ProductionConfig(BaseConfig):
db_name: str = "prod_db"
Question 1: Why should you avoid importing the main FastAPI instance inside route
modules? How do you organize routes instead?
Answer:
Importing the main FastAPI instance into sub-routers causes circular imports because the main app file
must import the routes to register them, and the route files would import the main app file to access the
app decorator.
To resolve this, we use APIRouter() to define routes in isolated sub-modules. The sub-modules do not
import app . Instead, we import the sub-routers and mount them to the main app instance in the central
[Link] entry point:
app.include_router(users_router) .
Question 2: How does Pydantic Settings help manage configuration keys safely?
Answer:
Pydantic Settings extends Pydantic's data validation to configuration management. It automatically
parses configuration keys from environment variables or .env files, validates their data types, raises
errors if mandatory keys are missing or invalid, and provides default values for optional configurations.
This ensures configuration errors are caught immediately at application startup.
13. Exercises
Define a Pydantic Settings configuration class containing two parameters: api_token (mandatory
string) and api_timeout (optional integer, defaults to 30). Set it up to load configurations from a local
.env file.
Solution:
class AppConfig(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
api_token: str
api_timeout: int = 30
This mini-project provides a complete, working skeleton of a modular FastAPI project. It implements a
config module, a mock database connection module, an isolated router, and the main app assembler.
# Simulated app/core/[Link]
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
app_name: str = "Modular Microservice Skeleton"
database_url: str = "sqlite:///[Link]"
settings = Settings()
# Simulated app/core/[Link]
from [Link] import settings
class FileDBProvider:
def __init__(self):
self.connection_string = settings.database_url
print(f"[DB] Initializing session with {self.connection_string}")
# Simulated app/api/v1/endpoints/[Link]
from fastapi import APIRouter, Depends
from [Link] import FileDBProvider, get_db_provider
from typing import Annotated
@[Link]("/")
async def list_users(db: Annotated[FileDBProvider, Depends(get_db_provider)]):
data = [Link]()
return {"users": ["alice", "bob"], "db_log": data}
# Simulated app/api/v1/[Link]
from fastapi import APIRouter
from [Link] import users
api_router = APIRouter()
# Register users sub-router under a clean prefix
api_router.include_router([Link], prefix="/users", tags=["Users"])
# Initialize application
app = FastAPI(title=settings.app_name)
@[Link]("/health")
async def health_check():
return {"status": "operational", "app": settings.app_name}
Key Takeaways
1. Single-file architectures do not scale. Apply the Separation of Concerns principle to decouple API
web code, service logic, and database persistence configurations.
2. Define routes using APIRouter() in sub-modules instead of importing the main FastAPI instance to
avoid circular imports.
3. Manage settings and secrets using Pydantic Settings ( BaseSettings ) to validate and load
configurations from environment variables and .env files safely.
4. Run your application using Python's module execution syntax ( python -m uvicorn [Link]:app )
from the root directory to ensure imports resolve correctly.
Further Reading
Architecture Patterns with Python by Harry Percival and Bob Gregory (O'Reilly)
1. Introduction
For most web applications, the database is the primary bottleneck. If your API endpoints process
requests in milliseconds but your database queries take seconds, your overall system latency will be
poor.
FastAPI applications frequently use PostgreSQL as their persistent relational database. Unlike MySQL
or SQLite, PostgreSQL is an enterprise-grade database designed for high-concurrency workloads, strict
data integrity, and complex queries.
However, to write efficient backend code, you must understand how PostgreSQL handles data
persistence under the hood. You cannot treat the database as a black box. You must understand:
In this chapter, we will explain PostgreSQL fundamentals from first principles, analyzing connection
mechanics, concurrency models, and SQL structures.
2. Theory
Relational databases store data in tables (relations) with strict columns and rows. Relationships
between tables are enforced using keys:
Atomicity: All operations in a transaction succeed, or the entire transaction is rolled back.
Consistency: Transactions bring the database from one valid state to another, enforcing
constraints.
Isolation: Concurrent transactions execute without interfering with each other.
Durability: Once a transaction is committed, it remains saved even in a power loss or crash.
The "I" in ACID (Isolation) defines how changes made by one transaction are visible to other concurrent
transactions. The SQL standard defines four isolation levels:
1. Read Uncommitted: A transaction can read uncommitted changes made by other transactions.
This allows dirty reads (reading data that is later rolled back). PostgreSQL does not support this
level and defaults to Read Committed instead.
2. Read Committed: A transaction only reads changes that have been committed. This prevents dirty
reads, but allows non-repeatable reads (if Transaction A reads a row, Transaction B updates that
row and commits, and Transaction A reads the row again, it gets the updated value). This is
PostgreSQL's default level.
3. Repeatable Read: Ensures that if a transaction reads data once, it gets the same data on
subsequent reads. This prevents non-repeatable reads, but allows phantom reads (Transaction A
queries a range of rows, Transaction B inserts a new row in that range and commits, and
Transaction A queries again, seeing the new "phantom" row). PostgreSQL's implementation of
Repeatable Read also prevents phantom reads.
4. Serializable: The strictest level. It simulates running transactions sequentially (one after another),
preventing all concurrency anomalies. However, it introduces significant locking overhead and can
cause transactions to fail with serialization errors, requiring your application to implement automatic
retry logic.
3. Internal Working
Unlike databases like MySQL (which allocate a thread per connection), PostgreSQL uses a process-
based model.
1. The main postmaster process forks a new helper process called a backend process to handle that
specific connection.
2. This backend process consumes operating system memory (typically 10MB to 20MB RAM) and
requires context switching at the CPU level.
Because processes are expensive to create and maintain, PostgreSQL has a hard limit on concurrent
connections (often configured using max_connections = 100 in [Link] ). If your web server
tries to open 500 concurrent connections, PostgreSQL will reject them, crashing your API.
[FastAPI Web Workers] ---> [Connection Pool] === (Shared TCP Connections) ===>
[PostgreSQL Backend Processes]
A Connection Pool (like PgBouncer or SQLAlchemy's built-in pool) maintains a fixed set of open TCP
connections to PostgreSQL. When a route handler needs a database session, it borrows a connection
from the pool, runs the query, and immediately returns the connection to the pool, allowing other
requests to reuse it.
4. API Reference
connect_timeout : The maximum seconds to wait for a socket connection to establish before failing.
keepalives : Periodically sends TCP keep-alive packets to prevent firewalls from dropping idle
connections.
5. Practical Examples
To understand database layers, we will use psycopg (version 3), the modern PostgreSQL adapter for
Python.
import psycopg
from [Link] import dict_row
try:
with [Link](conn_string, row_factory=dict_row) as conn:
# Open a transaction cursor
with [Link]() as cur:
# Create a table
[Link]("""
CREATE TABLE IF NOT EXISTS users (
id SERIAL PRIMARY KEY,
username VARCHAR(50) UNIQUE NOT NULL,
balance NUMERIC(10, 2) DEFAULT 0.00
);
""")
# Query record
[Link]("SELECT * FROM users WHERE username = %s;", ("alice",))
user_record = [Link]()
print(f"Retrieved User: {user_record}")
# Commit transaction automatically via context manager
except Exception as e:
print(f"Database error: {e}")
Never create a new connection connection directly inside an API endpoint. Use a connection pool
(configured via SQLAlchemy or an external utility like PgBouncer) and expose it as a dependency.
Limit your connection pool size to fit within PostgreSQL's max_connections limit across all active
web worker processes:
Pool Size = max_connections / number of web worker processes .
7. Common Mistakes
If you open a transaction inside a route handler but forget to commit or rollback the transaction before
exiting, that connection remains associated with the transaction, pinning locks on rows and preventing
other queries from executing.
Correction: Always use connection context managers ( with ) or clean yield dependency blocks to
guarantee rollbacks on failures.
8. Performance Tips
Avoid using SELECT * in your production queries. Fetching columns you do not use (like long text
descriptions or metadata JSON files) increases database memory utilization, consumes network
bandwidth, and prevents the database from leveraging index-only scan optimizations.
Best Practice: Specify exact column names in your queries: SELECT username, balance FROM
users; .
9. Security Considerations
Never connect your FastAPI web application to your database using the PostgreSQL superuser account
( postgres ).
Grant this user permissions only on the specific schemas and tables required for the application
( GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO app_user; ).
If your application freezes or times out, you may have run out of database connections. You can inspect
active PostgreSQL backend processes by running this query directly in the database:
This returns a list of active transactions, showing which queries are running and which client IP opened
them.
In high-throughput microservice architectures, you might scale your web server horizontally to run
dozens of containers. If each container opens a pool of 20 connections, you will exceed PostgreSQL's
connection limits.
Deploying PgBouncer (a lightweight connection pooler) in front of PostgreSQL allows thousands of web
containers to share a small, highly optimized pool of database connections, preventing connection
exhaustion.
Question 2: Explain the four SQL transaction isolation levels and identify PostgreSQL's
default.
Answer:
1. Read Uncommitted: Transactions can read uncommitted data, allowing dirty reads. (PostgreSQL
does not support this level).
2. Read Committed: Transactions can only read committed data, preventing dirty reads but allowing
non-repeatable reads. (PostgreSQL's default).
3. Repeatable Read: Guarantees that subsequent reads of the same rows within a transaction return
the same values, preventing non-repeatable reads. (PostgreSQL's implementation also prevents
phantom reads).
4. Serializable: Simulates sequential execution of transactions, preventing all anomalies at the cost of
high locking overhead.
13. Exercises
Write a raw SQL script to create two tables: users and orders . Enforce a foreign key relationship
between them, ensuring that if a user is deleted, their associated orders are also deleted automatically
( ON DELETE CASCADE ).
Solution:
This mini-project is a raw SQL balance ledger. It simulates transfering money between two accounts. It
runs the operations inside a transaction block, uses row-level locking ( SELECT ... FOR UPDATE ) to
prevent race conditions, and rolls back changes if a step fails.
import psycopg
from [Link] import dict_row
class BalanceLedger:
def __init__(self, conn_uri: str):
[Link] = conn_uri
def transfer_funds(self, from_user: str, to_user: str, amount: float) -> bool:
"""Executes a thread-safe funds transfer inside a transaction block."""
try:
with [Link]([Link], row_factory=dict_row) as conn:
with [Link]() as cur:
# 1. Fetch sender balance with a Row-Level Write Lock (FOR
UPDATE)
# This prevents other concurrent transactions from modifying
Alice's balance
[Link](
"SELECT balance FROM users WHERE username = %s FOR
UPDATE;",
(from_user,)
)
sender = [Link]()
if not sender:
raise ValueError(f"Sender '{from_user}' not found.")
# 4. Add to recipient
[Link](
"UPDATE users SET balance = balance + %s WHERE username =
%s;",
(amount, to_user)
)
Key Takeaways
1. Relational databases enforce ACID properties, ensuring reliable data persistence and transaction
isolation.
2. PostgreSQL utilizes a process-based connection model, making connection creation expensive.
Implement connection pooling to avoid resource exhaustion.
3. PostgreSQL's default transaction isolation level is Read Committed. Use row-level locking ( FOR
UPDATE ) inside transactions to prevent race conditions during read-modify-write loops.
4. Always use parameterized queries instead of string concatenation to protect your application from
SQL Injection attacks.
Further Reading
1. Introduction
Writing raw SQL queries (using database drivers like psycopg) is powerful, but it becomes cumbersome
as applications grow:
Writing database-agnostic queries is difficult; transitioning from SQLite (for testing) to PostgreSQL
(for production) requires rewriting queries.
An Object-Relational Mapper (ORM) solves this by mapping database tables directly to Python
classes. You interact with your database using standard object-oriented Python code.
In the Python ecosystem, SQLAlchemy is the leading ORM. It is split into two layers:
1. SQLAlchemy Core: A database abstraction layer that generates SQL queries programmatically.
In this chapter, we will master modern SQLAlchemy (version 2.0+), explore the internal session state
machine, and write database-independent schemas.
2. Theory
In modern SQLAlchemy 2.0, classes inherit from a base class that compiles them into database tables.
We use Python type hints ( Mapped[...] ) and mapped_column() to declare attributes:
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
Identity Map: Keeps track of all objects loaded from the database in a local registry inside the active
Session. If you query the same user row twice within a session, SQLAlchemy returns the exact
same object from memory, saving database query overhead.
Unit of Work: Instead of running database statements immediately, the Session registers changes
(inserts, updates, deletes) in memory. When you commit the session (or when a query is executed),
SQLAlchemy compiles and runs all updates in a single batch (flushing), minimizing TCP network
round-trips.
3. Internal Working
The most common source of bugs for developers is misunderstanding the state of objects inside an
active SQLAlchemy Session. An object exists in one of four states:
1. Transient: A newly created object in Python memory that is not associated with any session and
does not exist in the database (e.g., user = User(name="alice") ).
2. Pending: The object has been added to the session using [Link](user) . It is not yet written
to the database.
3. Persistent: The session has been flushed or committed. The object has a corresponding row in the
database and is tracked by the session.
4. Detached: The session has been closed or committed with database connections dropped. The
object remains in Python memory, but its connection to the database is severed. Accessing deferred
attributes on a detached object will raise a DetachedInstanceError crash.
4. API Reference
SQLAlchemy 2.0 replaced the legacy [Link]() syntax with core select statements:
Create: [Link](user)
Delete: [Link](user)
5. Practical Examples
This example establishes an in-memory SQLite engine, compiles the schema tables, and executes
standard CRUD operations.
# 2. Model Schema
class Product(Base):
__tablename__ = "products"
# Create tables
[Link].create_all(engine)
# READ
statement = select(Product).where([Link] == "Wireless Mouse")
product = [Link](statement).first()
print(f"Retrieved Product: {[Link]} - ${[Link]}")
# UPDATE
[Link] = 24.99 # Session tracks change automatically
[Link]()
# DELETE
[Link](product)
[Link]()
By default, SQLAlchemy expires object attributes when you call [Link]() . This means if you
read [Link] after committing, SQLAlchemy is forced to run a new SQL query to reload the
value.
7. Common Mistakes
Correction: Eager-load nested relationships inside the query, or read required attributes before exiting
the session boundary.
Slow Path:
[Link](insert(Product), items_list)
[Link]()
This generates a single bulk SQL statement, significantly reducing transaction round-trip latency.
9. Security Considerations
SQLAlchemy core compiles Select statements into parameterized queries dynamically. This ensures
that user inputs passed into filters are automatically bound as parameters (e.g. WHERE username =
:username ), completely isolating inputs and protecting your application from SQL Injection attacks.
To debug complex query mappings or check if indices are being utilized, enable SQL echo logging:
Alternatively, configure the Python root logger [Link] to redirect outputs to your log
aggregation pipelines.
Most production database structures require auditing attributes to trace data changes. Rather than
declaring created_at and updated_at on every model, define a reusable Mixin class.
Question 1: Explain the Session lifecycle states in SQLAlchemy. How do objects move
between them?
Answer:
1. Transient: Object is created in Python memory but not associated with a Session or written to the
database.
3. Persistent: Object is flushed to the database. It is tracked by the session, and has a valid primary
key.
4. Detached: The session is closed. The object remains in Python memory but changes are no longer
tracked, and lazy loading attributes raises errors.
Answer:
SQLAlchemy 1.x used a separate Query API ( [Link](Model) ) that was distinct from Core's
SQL builder syntax.
SQLAlchemy 2.0 unified the query engine, making select() the standard interface for both Core and
ORM. This allows queries to be compiled using standard syntax, improves static type-hinting support in
IDEs, and enables better execution caching under the hood.
13. Exercises
Define a model Task with attributes: id (primary key), title (string, max 100, nullable=False),
is_completed (boolean, default=False), and mix in AuditMixin .
Solution:
This mini-project implements a complete database service class. It configures the engine and session
factory, handles session contexts, manages audit mixins, and exposes a clean, standardized CRUD
interface for a database model.
# 1. Base Declaration
class Base(DeclarativeBase):
pass
class AuditMixin:
created_at: Mapped[datetime] = mapped_column(default=[Link]())
updated_at: Mapped[datetime] = mapped_column(default=[Link](),
onupdate=[Link]())
# 2. Database Model
class Customer(Base, AuditMixin):
__tablename__ = "customers"
@contextmanager
def get_session(self):
"""Yields transactional database sessions with automatic rollbacks."""
session = self.session_factory()
try:
yield session
[Link]()
except Exception:
[Link]()
raise
finally:
[Link]()
customer = Customer(email=email)
[Link](customer)
# Customer moves from Pending -> Persistent upon session commit inside
context
return customer
Key Takeaways
1. Object-Relational Mappers (ORMs) map database tables to Python classes, improving code
maintainability.
2. Modern SQLAlchemy 2.0 uses Python type hints ( Mapped[...] ) and select() statement queries
to align ORM structures with Python standards.
3. Understand the Session lifecycle states: Transient, Pending, Persistent, and Detached. Do not read
relationships on detached objects.
4. Set expire_on_commit=False inside the sessionmaker configuration to prevent unnecessary
database queries when accessing attributes after committing a transaction.
Further Reading
1. Introduction
FastAPI's strength is handling thousands of concurrent requests on a single thread using asynchronous
I/O. However, in traditional database layers, database operations block. If your route handler calls
[Link]() on a synchronous SQLAlchemy session, the entire event loop blocks, waiting for the
database server to process the update.
To solve this, SQLAlchemy introduced native support for asynchronous execution. By combining:
Async Database Drivers (like asyncpg for PostgreSQL or aiosqlite for SQLite).
Your database queries can be executed asynchronously, allowing the event loop to switch to other
requests while waiting for the database to return data.
In this chapter, we will learn how to configure and run Async SQLAlchemy, integrate database sessions
with FastAPI's dependency injection system, and handle common async-persistence pitfalls.
2. Theory
The Synchronous Drivers Conflict
When a database query runs, the database client library opens a TCP socket and writes data. In a
synchronous driver (such as psycopg2 ), the socket write operation is blocking. The calling Python
thread is put to sleep by the OS until the server responds.
An asynchronous driver (such as asyncpg ) uses non-blocking sockets. It registers the database query
write event on the active event loop, yields control, and resumes execution only when the loop notifies it
that database packets have arrived.
Because SQLAlchemy's core was designed around synchronous design patterns, it uses a utility called
greenlet to enable async support.
A greenlet is a lightweight, in-process coroutine context. SQLAlchemy uses greenlet to intercept internal
database queries (like resolving lazy-loaded attributes) and translate them into async operations behind
the scenes, allowing you to use standard ORM patterns without violating async boundaries.
3. Internal Working
When you call await [Link](statement) , the session uses the async_engine to write data.
The execution context yields control back to the event loop.
AsyncSession ---> [Greenlet Adapter] ---> [asyncpg socket write] ---> (Event Loop
yields control)
Because of this greenlet layer, if your code attempts to access a database model relation that has not
been eagerly loaded (which triggers a query), and the active database session is closed or doesn't
support greenlet task context propagation, SQLAlchemy raises a MissingGreenlet exception.
4. API Reference
engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = async_sessionmaker(
bind=engine,
expire_on_commit=False
)
5. Practical Examples
This example configures an async engine using an in-memory SQLite database and executes async
queries.
import asyncio
from sqlalchemy import select
from [Link] import create_async_engine, async_sessionmaker,
AsyncAttrs
from [Link] import DeclarativeBase, Mapped, mapped_column
# 2. Schema
class Article(Base):
__tablename__ = "articles"
# READ
statement = select(Article).where([Link]("%Async%"))
result = await [Link](statement) # Await execute!
article = [Link]().first()
print(f"Retrieved Article: {[Link]}")
await [Link]()
[Link](main())
To use database sessions in your endpoints, create a dependency that yields an AsyncSession per
request. Use asynccontextmanager to ensure the session is always closed, even if the request fails.
7. Common Mistakes
Mistake: Accessing relationships that have not been loaded from the database inside an async context.
# Assuming User has a relationship to Orders
result = await [Link](select(User))
user = [Link]().first()
# Accessing [Link] triggers a synchronous lazy-load query, raising
MissingGreenlet!
print([Link])
Correction: Always eagerly load relationships using selectinload or joinedload options inside your
query:
statement = select(User).options(selectinload([Link]))
result = await [Link](statement)
user = [Link]().first()
print([Link]) # Safe. Loaded in the initial query.
Async engines maintain active socket pools in the background. If you restart or terminate your
application without calling await [Link]() , database connections remain open, leading to
connection leaks.
Correction: Register a clean disposal call in your FastAPI application's lifespan shutdown handler.
8. Performance Tips
In highly concurrent async applications, set the connection pool size ( pool_size ) and max overflow
limit ( max_overflow ) when creating the engine to prevent requests from queueing up:
engine = create_async_engine(
DATABASE_URL,
pool_size=20, # Maximum persistent database connections
max_overflow=10, # Temporary overflow connection limit
pool_timeout=30 # Seconds to wait for a free connection before failing
)
9. Security Considerations
If an endpoint queries a slow database table and takes several seconds to return, it holds onto its
database connection from the pool. If a large volume of requests targets this route, the connection pool
will be exhausted. Any subsequent request will time out, causing a Denial of Service (DoS) crash.
Mitigation: Add strict execution timeouts to queries, configure database read replicas, and scale
your pool limits appropriately.
If your application raises MissingGreenlet errors, check your query definitions. Ensure that all nested
attributes referenced in serialization schemas (like Pydantic response models) are eagerly loaded in the
query.
A common real-world scenario is fetching data concurrently from both a database and an external third-
party API inside an endpoint. Using an async session allows both operations to run concurrently.
Answer:
The MissingGreenlet exception occurs when you attempt to access a database model relationship that
has not been eagerly loaded (which triggers a lazy-load query) inside an async environment. Because
async database calls must be explicitly awaited, accessing a relationship attribute dynamically fails
because attribute access in Python is synchronous.
To resolve this, you must eagerly load the relationship in the initial query using selectinload (for one-
to-many relationships) or joinedload (for many-to-one relationships).
Answer:
create_async_engine requires asynchronous drivers because standard drivers (like sqlite3 or
psycopg2 ) perform blocking socket operations. If a sync driver is used, the main event loop thread
blocks during database operations, nullifying the benefits of FastAPI's async architecture. Async drivers
use non-blocking socket operations, allowing the event loop to process other concurrent requests while
waiting for the database to respond.
13. Exercises
Solution:
This mini-project is a fully async user manager database service. It defines an async SQLite database,
integrates it with FastAPI's dependency injection system, and implements non-blocking registration and
query endpoints.
class DBUser(Base):
__tablename__ = "users"
app = FastAPI(lifespan=lifespan)
# 4. Pydantic Schemas
class UserRegister(BaseModel):
username: str
email: EmailStr
class UserResponse(BaseModel):
id: int
username: str
email: str
class Config:
from_attributes = True
@[Link]("/users/{username}", response_model=UserResponse)
async def get_user(username: str, db: AsyncDB):
# Non-blocking query
stmt = select(DBUser).where([Link] == username)
res = await [Link](stmt)
user = [Link]().first()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
Key Takeaways
1. Standard database drivers block the event loop. Always use async drivers (like asyncpg or
aiosqlite ) and async engines ( create_async_engine ) in async applications.
2. SQLAlchemy uses greenlet to run database calls asynchronously under the hood.
4. Inject database sessions into routes using FastAPI dependency providers, and ensure sessions are
always closed safely.
Further Reading
1. Introduction
When you deploy a database-backed web application to production, you cannot compile tables from
scratch using [Link].create_all(engine) every time you change your code. Doing so would
wipe out all existing production database tables and data.
Instead, database schemas must evolve incrementally. As requirements change, you might need to add
a column, modify a data type, delete a table, or migrate existing records.
Alembic is the standard lightweight database migration tool for SQLAlchemy. It acts like Git for your
database. It tracks changes to your SQLAlchemy models, generates chronological SQL migration
scripts, applies modifications incrementally using version tables, and supports rollbacks to previous
states.
In this chapter, we will learn how to configure Alembic, set up support for asynchronous engines, auto-
generate migration files, and deploy migrations safely.
2. Theory
What is a Migration?
upgrade() : The SQL operations needed to transition the database schema to the next version (e.g.,
adding a phone column).
downgrade() : The operations needed to revert those changes if the migration fails or needs to be
rolled back (e.g., dropping the phone column).
When you initialize Alembic, it creates a special table in your database named alembic_version
containing a single column and row: version_num .
This row stores the unique identifier (hash) of the active migration script. When you run migrations,
Alembic reads this hash, compares it with your local migration scripts, and executes only the missing
versions.
3. Internal Working
Alembic's autogenerate command compares your database's actual schema (tables and columns) with
your active SQLAlchemy Python model classes:
[SQLAlchemy Models (Python)] <--- (Alembic compares) ---> [Actual Database Schema]
|
v
Generates migration revision file
1. It checks which tables are defined in SQLAlchemy but missing in the database (generating
create_table commands).
2. It checks which columns have been added, removed, or modified in your classes (generating
add_column or drop_column commands).
3. It writes these changes into a new migration file inside your versions/ folder.
It does not detect index changes or custom checks by default unless configured.
It does not detect table renames (it will generate a drop_table and create_table command
instead, resulting in data loss if run!). Always review auto-generated scripts before applying
them.
4. API Reference
alembic upgrade head : Applies all pending migrations up to the latest revision.
5. Practical Examples
import asyncio
from [Link] import fileConfig
from sqlalchemy import pool
from [Link] import create_async_engine
from alembic import context
await [Link]()
if context.is_offline_mode():
run_migrations_offline()
else:
# Run async loop
[Link](run_migrations_online())
Never run alembic upgrade head in production with raw, unchecked auto-generated scripts.
Open the generated Python revision file inside your versions/ directory.
Add custom data migration logic if you are renaming columns or splitting tables to avoid data loss.
Run migrations before the new application containers boot up to ensure the database schema is
updated before routes receive traffic.
7. Common Mistakes
Mistake: Running alembic revision --autogenerate and seeing that Alembic generates empty
upgrade() and downgrade() blocks, even though you added new models.
Correction: Alembic does not automatically scan your codebase for models. You must explicitly import
all model modules (files) inside migrations/[Link] so the Python interpreter registers them on
[Link] before Alembic runs:
# In [Link]
from [Link].db_models import User, Item, Order # Critical!
Mistake: Hardcoding database credentials in the [Link] line of the [Link] file.
Correction: Keep [Link] credentials-free and load the database URL dynamically inside [Link]
from environment variables:
# In [Link]
import os
config.set_main_option("[Link]", [Link]["DATABASE_URL"])
8. Performance Tips
Migrations are run as isolated, short-lived script executions. When configuring the database engine
inside [Link] , use NullPool to bypass connection pooling overhead:
This prevents Alembic from occupying open connections in database sockets after the migration scripts
complete.
9. Security Considerations
If you delete a column containing sensitive customer data (like passwords or credit card numbers),
double-check that the generated migration code does not copy or store this data in temporary backup
tables inside the database.
If two developers create migrations locally at the same time and push their code, the git merge will
result in a split-head conflict (the migration tree has two "latest" versions, and Alembic does not know
which to run).
Fix: Run alembic heads to view the conflict hashes. Merge the branches into a single head by
running:
alembic merge [hash1] [hash2] -m "merge split heads"
This generates a new migration file linking both paths, resolving the conflict.
To ensure database migrations succeed without errors, developers run a test database container in their
pull-request pipelines, apply migrations from scratch, and verify that the database table schema
matches the local code models.
Question 1: How does Alembic track which migrations have already been run on a
database?
Answer:
Alembic creates a table named alembic_version in the database. This table contains a single column
version_num containing a single row that stores the unique revision hash of the most recently executed
migration. When alembic upgrade head is run, Alembic reads this hash, looks up the corresponding
migration script in the codebase, and executes only the subsequent migration scripts in chronological
order.
Question 2: Why must you import your database model files inside migrations/[Link] ?
Answer:
Alembic compares the database schema against the metadata registry ( [Link] ) of your
SQLAlchemy models. If you do not import your model files inside [Link] , Python does not execute the
classes, and the models are not registered in the [Link] collection. As a result, Alembic will
assume these models do not exist and will generate scripts to delete your database tables.
13. Exercises
Write a migration upgrade function that manually updates a status column from NULL to a default
string "pending" using raw SQL execution statements.
Solution:
def upgrade():
# Update existing null columns safely before adding strict constraints
[Link]("UPDATE orders SET status = 'pending' WHERE status IS NULL;")
This mini-project demonstrates how to define a customized migration revision script with safety
assertions and transactional isolation overrides.
"""
Alembic migration revision template.
Revision ID: a1b2c3d4e5f6
"""
revision = 'a1b2c3d4e5f6'
down_revision = None
branch_labels = None
depends_on = None
from alembic import op
import sqlalchemy as sa
def upgrade():
# 1. Create a secure 'users' table
op.create_table(
'users',
[Link]('id', [Link](), primary_key=True),
[Link]('username', [Link](length=50), nullable=False, unique=True),
[Link]('email', [Link](length=100), nullable=False),
[Link]('is_verified', [Link](), server_default=[Link]('false'),
nullable=False)
)
def downgrade():
# Drop index and table in reverse order
op.drop_index('idx_users_username')
op.drop_table('users')
Key Takeaways
1. Alembic operates as a database schema version control tool, tracking database evolution using an
alembic_version table.
2. The alembic revision --autogenerate command compares code models with the database
schema to generate migrations, but it cannot detect all changes (such as index modifications or
column renames) and should always be reviewed.
3. Configure migrations/[Link] to use an async engine and NullPool connection pool to execute
migrations cleanly in async environments.
4. Integrate migrations ( alembic upgrade head ) into your CI/CD deployment pipelines to update
schemas automatically before the application handles user traffic.
Further Reading
1. Introduction
Relational databases represent connections between entities using keys. For example, a user has many
posts, and an order contains many products.
In raw SQL, you query these relationships by writing JOIN queries. In SQLAlchemy, you represent
these relationships using the relationship directive. This allows you to navigate related objects
dynamically using standard attribute access:
However, if you do not configure how these relationships are loaded, SQLAlchemy can introduce severe
performance bottlenecks, such as the N+1 Query Problem.
In this chapter, we will learn how to map one-to-many, many-to-one, and many-to-many relationships,
configure cascade rules, select the correct loading strategy, and write efficient join queries.
2. Theory
Relationship Types
1. One-to-Many (O2M): A parent row is associated with multiple child rows (e.g. one User has many
Posts).
2. Many-to-One (M2O): Multiple rows reference a single target row (e.g. many Posts belong to one
User).
3. Many-to-Many (M2M): Rows in Table A can relate to multiple rows in Table B, and vice-versa (e.g.
many Posts have many Tags). This requires a third table called an association table to store the
mapping.
Loading Strategies
Lazy Loading ( lazy="select" ): SQLAlchemy only queries the database for related data when you
access the attribute. This is the default setting but causes the N+1 query problem in lists.
Joined Eager Loading ( joinedload ): Fetches the parent and related rows in a single SQL
statement using a LEFT OUTER JOIN . Best for many-to-one relationships.
Select IN Eager Loading ( selectinload ): Runs two SQL statements: one to fetch the parents, and
a second SELECT ... WHERE parent_id IN (...) statement to fetch all related children. Best for
one-to-many relationships.
Raise Loading ( lazy="raise" ): Raises an error if your code attempts to access the attribute
without eager loading it first. Excellent for catching lazy loading bugs during development.
3. Internal Working
The N+1 query problem occurs when you fetch a list of parent rows and loop over them, accessing a
lazy-loaded relationship for each parent:
If you have 100 users, this executes 101 database queries (1 + 100), causing severe server latency.
Using selectinload resolves this, reducing the operation to exactly 2 queries (1 + 1), regardless of
how many users are in the list.
4. API Reference
"all, delete-orphan" : If the parent is deleted, delete all children. If a child is removed from
the parent list, delete the orphaned child row from the database.
5. Practical Examples
One-to-Many & Many-to-Many Mappings
This example configures a One-to-Many relationship (User -> Posts) and a Many-to-Many relationship
(Post -> Tags) using an association table.
class Base(DeclarativeBase):
pass
# 2. Tag Model
class Tag(Base):
__tablename__ = "tags"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(unique=True, nullable=False)
# 3. Post Model
class Post(Base):
__tablename__ = "posts"
id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str] = mapped_column(nullable=False)
user_id: Mapped[int] = mapped_column(ForeignKey("[Link]",
ondelete="CASCADE"))
# 4. User Model
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(unique=True, nullable=False)
To prevent N+1 query bugs from reaching production, configure your relationships to raise errors if
accessed without eager loading.
Add lazy="raise" to your relationships in development. This forces unit tests to crash if an
endpoint attempts to lazy load data, allowing you to catch and fix query issues early.
7. Common Mistakes
Correction: Use back_populates . It requires defining relationships explicitly on both models, enabling
accurate static type analysis and IDE auto-completion.
8. Performance Tips
Use joinedload for Many-to-One relations (like fetching a Post and its Author), because a single
SQL join is highly optimized for fetching single parent records.
Use selectinload for One-to-Many or Many-to-Many relations, because SQL joins on collections
create duplicate rows in the result set, increasing network overhead and memory parsing time.
9. Security Considerations
If a Pydantic model serializes a User including their Posts, and the Post schema serializes the Author
(User), this creates an infinite loop, crashing your application with a stack overflow error.
Mitigation: Create dedicated schemas for nested models and exclude circular references:
class PostResponse(BaseModel):
id: int
title: str
class UserResponse(BaseModel):
id: int
username: str
posts: list[PostResponse] # PostResponse does not include the author
To verify your loading strategies, use python's echo=True or count query statements inside your unit
tests to ensure that fetching a list of records runs exactly 1 or 2 SQL queries rather than scaling with the
number of records.
Self-referential relationships are used to build hierarchies like comment threads, where a comment can
have child replies:
class Comment(Base):
__tablename__ = "comments"
id: Mapped[int] = mapped_column(primary_key=True)
parent_id: Mapped[int | None] = mapped_column(ForeignKey("[Link]",
ondelete="CASCADE"))
# Self-referential relationship
replies: Mapped[List["Comment"]] = relationship(
cascade="all, delete-orphan",
back_populates="parent"
)
parent: Mapped["Comment"] = relationship(back_populates="replies", remote_side=
[id])
Question 1: What is the N+1 query problem, and how does selectinload solve it?
Answer:
The N+1 query problem occurs when you load a parent list (1 query) and loop over it, triggering a lazy-
loaded database query for each parent's relationship (N queries). This results in N+1 total queries.
selectinload solves this by running exactly two queries: the first query fetches the parent list, and the
second query executes a SELECT ... WHERE parent_id IN (...) statement to fetch all related child
records in a single batch, reducing query overhead to a constant 2.
Answer:
joinedload uses a SQL LEFT OUTER JOIN to fetch the parent and related rows in a single query. It
is best for Many-to-One relations (e.g. Post to Author) because it retrieves the associated row in one
step.
selectinload runs a second query using an IN clause to fetch the child rows in a batch. It is best
for One-to-Many (User to Posts) or Many-to-Many relations, as using joins on large collections
results in a Cartesian product (duplicated rows in the result set), which degrades database and
application performance.
13. Exercises
Map a many-to-many relationship between Student and Course using an association table named
enrollments .
Solution:
enrollment_association = Table(
"enrollments",
[Link],
Column("student_id", ForeignKey("[Link]", ondelete="CASCADE"),
primary_key=True),
Column("course_id", ForeignKey("[Link]", ondelete="CASCADE"),
primary_key=True)
)
class Student(Base):
__tablename__ = "students"
id: Mapped[int] = mapped_column(primary_key=True)
courses: Mapped[list["Course"]] =
relationship(secondary=enrollment_association)
class Course(Base):
__tablename__ = "courses"
id: Mapped[int] = mapped_column(primary_key=True)
This mini-project is a database schema for an e-commerce platform. It maps Customers, Orders, and
Items, configures delete-cascade rules, and provides an Order Service to fetch orders and their line
items in a single step using eager loading.
class Base(DeclarativeBase):
pass
# 2. Item Model
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True)
sku: Mapped[str] = mapped_column(unique=True)
price: Mapped[float] = mapped_column()
# 3. Order Model
class Order(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True)
customer_id: Mapped[int] = mapped_column(ForeignKey("[Link]",
ondelete="CASCADE"))
# Many-to-One to Customer
customer: Mapped["Customer"] = relationship(back_populates="orders")
# Many-to-Many to Items
items: Mapped[List[Item]] = relationship(secondary=order_items)
# 4. Customer Model
class Customer(Base):
__tablename__ = "customers"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
Key Takeaways
1. Map relationships using relationship() , linking models with back_populates on both classes to
ensure consistent state.
2. Lazy loading ( lazy="select" ) is the default loading strategy, but it causes N+1 query problems in
loops.
3. Resolve the N+1 query problem using eager loading: selectinload for collections (one-to-
many/many-to-many) and joinedload for single entities (many-to-one).
4. Configure cascade rules ( cascade="all, delete-orphan" ) on parent models to delete related child
records automatically, preventing orphaned rows.
Further Reading
1. Introduction
When multiple users query and update the same data concurrently, race conditions occur. A classic
example is a ticket booking system:
If your application does not implement concurrency controls, both users will pay for the same ticket,
resulting in a double booking.
To prevent this, you must control how database transactions read and write shared data using locking.
In this chapter, we will master database concurrency control patterns, learn the differences between
optimistic and pessimistic locking, and implement thread-safe transaction blocks using SQLAlchemy.
2. Theory
Pessimistic Locking
Pessimistic locking assumes conflicts are highly likely. It prevents conflicts by locking the rows during a
read operation, forcing other transactions to block (wait) until the lock is released.
Optimistic Locking
Optimistic locking assumes conflicts are rare. It allows transactions to read and update data without
acquiring locks. However, it adds a version or timestamp column to the table. When updating:
1. The transaction reads the row and records the current version (e.g., version = 1 ).
3. If another transaction updated the row in the meantime, the version is no longer 1 . The update
affects 0 rows, signaling to the application that a conflict occurred.
Cons: The application must catch conflicts and implement retry logic.
Deadlocks
A deadlock occurs when two transactions are blocked, each waiting for a lock held by the other.
Both transactions freeze indefinitely. PostgreSQL detects this and terminates one of the transactions,
raising an error.
Deadlock Prevention: Always acquire locks in the same order (e.g., order by primary key ID) in all
transactions.
3. Internal Working
When you run SELECT ... FOR UPDATE in PostgreSQL, the database acquires a RowShareLock on the
matched rows.
If Transaction B attempts to run SELECT ... FOR UPDATE on the same rows, it is blocked.
If Transaction B attempts to run a standard SELECT (without locks), it is not blocked (PostgreSQL
uses Multi-Version Concurrency Control (MVCC) to return the last committed state).
4. API Reference
nowait=True : Raises a database error immediately if the lock is held, rather than waiting.
with_for_update(nowait=True)
skip_locked=True : Skips locked rows and returns only unlocked rows. Excellent for message
queues or worker pools.
with_for_update(skip_locked=True)
5. Practical Examples
class Ticket(Base):
__tablename__ = "tickets"
id: Mapped[int] = mapped_column(primary_key=True)
is_booked: Mapped[bool] = mapped_column(default=False)
# SQLAlchemy's version_id_col automatically tracks version changes
version: Mapped[int] = mapped_column(default=1)
__mapper_args__ = {
"version_id_col": version
}
ticket.is_booked = True
try:
await [Link]()
return True
except StaleDataError:
# A collision occurred! Revert changes.
await [Link]()
return False
Do not place long-running, non-database operations (like calling external APIs, writing files, or sleeping)
inside transaction blocks.
Keep connection sessions open for the minimum time necessary to minimize lock duration and
prevent connection pool starvation.
7. Common Mistakes
8. Performance Tips
If you are building a background task processor where multiple workers query a table to pick up task
records:
# Fetches the next available task, skipping rows already being processed
statement = (
select(TaskQueue)
.where([Link] == "pending")
.limit(1)
.with_for_update(skip_locked=True)
)
9. Security Considerations
1. Locking-based DoS
If an API allows a user to lock rows using a query parameter without restrictions, an attacker could open
a transaction, lock critical tables using SELECT ... FOR UPDATE , and leave the connection open,
freezing the application for all users.
Mitigation: Never expose raw lock targets to client query parameters. Add strict query execution
timeouts to prevent connections from remaining open.
Check your application logs to see which queries triggered the deadlock.
Ensure all endpoints acquire locks in the exact same table sequence (e.g. always query Users
before Accounts ).
11. Real-world Use Cases
Pessimistic locking is the industry standard for financial accounts, ticket booking, and hotel reservations
where double-booking cannot be tolerated under any circumstances.
Question 1: What is the difference between optimistic and pessimistic locking? When
should you use each?
Answer:
Pessimistic Locking acquires database locks when reading rows (using FOR UPDATE ), blocking
other transactions from modifying those rows until the transaction commits. Use it when conflict
rates are high (e.g. ticket sales or inventory checks).
Optimistic Locking does not acquire locks. It uses a version column to check if the data changed
before writing. Use it when conflict rates are low (e.g. user profile updates) to maximize
performance.
Answer:
In background worker queues, multiple workers query the same table to fetch pending tasks. Without
skip_locked=True , a worker locks a row using FOR UPDATE , forcing other workers to block on that row.
By using skip_locked=True , other workers skip the locked row and process the next available task,
allowing all workers to operate concurrently without blocking.
13. Exercises
Write a query using SQLAlchemy select() that fetches an Account by ID and locks the row using
nowait=True .
Solution:
statement = (
select(Account)
.where([Link] == account_id)
.with_for_update(nowait=True)
)
14. Mini Project: Thread-Safe Room Booking Service
This mini-project is a room booking service. It defines an async room model, implements pessimistic
locking ( with_for_update ) to prevent double bookings, and handles locking timeouts gracefully.
class Base(DeclarativeBase):
pass
class Room(Base):
__tablename__ = "rooms"
class BookingService:
@staticmethod
async def reserve_room(session: AsyncSession, room_number: str) -> dict:
"""Reserve a room safely using pessimistic locking with nowait."""
try:
# 1. Fetch Room and lock the row using nowait=True
statement = (
select(Room)
.where(Room.room_number == room_number)
.with_for_update(nowait=True) # Fail fast if another worker is
updating
)
result = await [Link](statement)
room = [Link]().first()
if not room:
return {"status": "error", "message": "Room not found."}
if room.is_reserved:
return {"status": "error", "message": "Room is already reserved."}
# 2. Complete reservation
room.is_reserved = True
await [Link]()
return {"status": "success", "message": f"Room {room_number} reserved
successfully."}
Key Takeaways
1. Concurrency control prevents race conditions (like double bookings) when multiple users access
shared data.
2. Pessimistic locking locks rows using SELECT ... FOR UPDATE , forcing other transactions to wait.
3. Optimistic locking uses a version column to detect conflicts at write time, offering high performance
when conflict rates are low.
4. Keep transaction blocks short to minimize lock holding times and prevent database connection pool
timeouts.
Further Reading
Database Internals by Alex Petrov (O'Reilly - for transactional engines and locking details)
1. Introduction
As your database grows from thousands of rows to millions of rows, queries that used to run instantly
will begin to slow down. If a query scans a table sequentially (checking every single row in the table to
find a match), its execution time scales linearly with the size of the database.
However, indexes are not free. They slow down write operations (inserts, updates, deletes) because the
database must update the index structures alongside the table rows.
In this chapter, we will learn how indexing works, how to analyze query performance using EXPLAIN
ANALYZE , and how to design optimal composite indexes.
2. Theory
B-Tree Indexes
By default, PostgreSQL creates B-Tree (Balanced Tree) indexes. A B-Tree index maintains data in a
sorted, balanced tree structure, allowing the database to perform binary-search-like operations:
[ Root Node: 50 ]
/ \
[ Branch Node: 25 ] [ Branch Node: 75 ]
/ \ / \
[ Leaf: 10, 20 ] [ Leaf: 30 ] [ Leaf: 60 ] [ Leaf: 80, 90 ]
To find a record with ID 75 , the database starts at the root, moves to the right branch (since 75 > 50),
and reaches the leaf node containing the pointer to the physical disk location of row 75 . This lookup
takes O(log N ) time instead of O(N ) sequential scanning time.
A Composite Index is an index built on multiple columns (e.g., INDEX (last_name, first_name) ).
The Leftmost Rule: A composite index can only be used by queries that filter by columns starting
from the left.
A query filtering by last_name or last_name AND first_name will use the index.
3. Internal Working
To optimize queries, you must ask PostgreSQL how it plans to execute them.
EXPLAIN : Returns the execution plan compiled by the query optimizer based on statistical metadata,
without running the query.
EXPLAIN ANALYZE : Executes the query and returns the actual execution times, loop counts, and
memory allocations.
Sequential Scan (Seq Scan): The database scans the entire table on disk. This is slow for large
tables.
Index Scan: The database reads the index to find row pointers, then fetches the actual rows from
the table disk space.
Index Only Scan: The database finds the requested data directly inside the index itself and avoids
fetching the table rows from disk. This is the fastest scan type.
4. API Reference
You declare indexes directly inside your model class definition using the Index class:
class Employee(Base):
__tablename__ = "employees"
# Index definitions
__table_args__ = (
Index("idx_employee_email", "email"), # Single column index
Index("idx_emp_dept", "department_id", "email") # Composite index
)
5. Practical Examples
This example illustrates how to run a query trace to analyze database execution scans.
High Cardinality: Emails, UUIDs, usernames (almost every row has a unique value). Excellent
targets for indexing.
Low Cardinality: Status (e.g. pending , completed ), gender, active flags. Bad targets for B-Tree
indexing.
If a column only has 3 possible values, a query filtering by that column will match a large percentage
of the table. In this case, PostgreSQL will bypass the index and perform a Sequential Scan because
scanning the table directly is faster than reading the index and fetching table pages.
7. Common Mistakes
Mistake: Adding an index to every column in a table to try and optimize all potential queries.
Every index slows down database writes (inserts, updates, deletes) because the database must update
the index alongside the row. It also increases storage requirements.
Correction: Only index columns that are frequently used in WHERE filters, JOIN conditions, or ORDER
BY sorting clauses.
Mistake: Creating a composite index on (country, state, city) but running queries that filter only by
state . This query will bypass the index and perform a Seq Scan.
Correction: Ensure the query filters start with the leftmost column of the index, or create separate
single-column indexes.
8. Performance Tips
If you require high throughput for a query (e.g. looking up a user's verification status), select only the
indexed columns:
9. Security Considerations
Ensure that any query executed by your API endpoints uses an index. If an unindexed query is exposed
to users, a malicious actor could send thousands of concurrent requests to that route, forcing the
database to perform a Sequential Scan for each request. This exhausts database CPU, causing a
system-wide outage.
Configure PostgreSQL to log queries that take longer than a specified threshold (e.g., 200ms) by
modifying [Link] :
This writes slow queries to your system logs, helping you identify which tables require indexing.
If a table contains millions of soft-deleted records ( is_deleted=True ), you can create a Partial Index
that only indexes active rows, keeping the index size small:
Question 1: How does a B-Tree index work? Why is search time O(log N) instead of
O(N)?
Answer:
A B-Tree index organizes data in a sorted, balanced tree structure. The database starts at the root
node, compares values, and navigates down to the leaf node containing the address pointer of the
target row. Because the tree is sorted and balanced, search time is logarithmic O(log N ) instead of
linear O(N ) (sequential scanning).
Answer:
A composite index (an index on columns A, B, and C) acts as a sorted index. It is sorted primarily by
column A, then by column B, then by column C. Because of this sorting order, a query must filter by
column A first to use the index. A query filtering only by B or C cannot use the index, and the database
will perform a sequential scan.
13. Exercises
Define a SQLAlchemy model Order with customer_id and status columns, and add a composite
index on both fields.
Solution:
class Base(DeclarativeBase):
pass
class Order(Base):
__tablename__ = "orders"
__table_args__ = (
Index("idx_customer_status", "customer_id", "status"),
)
This mini-project demonstrates how to measure query execution times on indexed vs. unindexed
columns using an in-memory database.
import time
from sqlalchemy import create_engine, select, Index
from [Link] import DeclarativeBase, Mapped, mapped_column, sessionmaker
class Base(DeclarativeBase):
pass
class LogRecord(Base):
__tablename__ = "log_records"
__table_args__ = (
Index("idx_log_trace", "trace_id"), # Index on trace_id
)
Key Takeaways
1. Indexes organize table columns in a sorted B-Tree structure, reducing search latency from linear
O(N ) to logarithmic O(log N ) scale.
2. Every index slows down write operations. Avoid indexing low-cardinality columns.
3. Use composite indexes for queries filtering by multiple columns, keeping in mind the leftmost rule.
4. Run queries with EXPLAIN ANALYZE directly in the database to inspect cost estimations and scan
types (Seq Scan, Index Scan, Index Only Scan).
Further Reading
1. Introduction
When an API exposes an endpoint to list records (e.g. /api/v1/items ), you must never return all rows
from the database. If your table contains millions of rows, returning them in a single response will:
Pagination is the practice of splitting a dataset into distinct, manageable chunks (pages) that are
returned to the client on demand.
1. Offset-based Pagination: Fetching data using limits and offsets (e.g. "page 3").
2. Cursor-based (Keyset) Pagination: Fetching data using a reference pointer from the last page
(e.g. "fetch 10 records after ID 42").
In this chapter, we will analyze the trade-offs of both approaches, explore database performance
characteristics, and implement clean pagination APIs in FastAPI.
2. Theory
Offset-based Pagination
Offset pagination uses two parameters: limit (page size) and offset (number of rows to skip).
Pros:
Simple to implement.
Allows users to jump to an arbitrary page (e.g., page 15).
Cons:
Performance degradation: To return data at OFFSET 1000000 , the database must scan and
sort all preceding 1,000,000 rows on disk before discarding them, leading to slow queries on
deep pages.
Data Drift: If a new item is inserted or deleted while a user is scrolling, the items will shift,
causing the user to see duplicate items or miss items entirely.
Cursor pagination uses a pointer (the "cursor") referencing a specific record in the dataset (usually the
ID or a timestamp). The next page is fetched relative to this cursor.
SQL Query: SELECT * FROM items WHERE id > 42 ORDER BY id LIMIT 10;
Pros:
Scale Invariant: The database performs a quick index lookup to find ID 42 and returns the next
10 rows. Search performance is constant O(1) regardless of depth.
No Data Drift: Because the query is bound to a specific ID, adding or deleting items does not
duplicate or skip items.
Cons:
Cursor-based:
Page 1: [1 ... 10] (Cursor: 10) ---> Page 2: WHERE id > 10 LIMIT 10 ---> Page 3:
WHERE id > 20 LIMIT 10
3. Internal Working
Offset Scan: When running OFFSET 5000 , PostgreSQL reads the index, traverses to the disk heap,
loads the rows, counts them, and discards them. This creates significant disk I/O overhead.
Cursor Index Scan: Running WHERE id > 5000 LIMIT 10 leverages the B-Tree index to jump
directly to leaf node 5000 and reads the next 10 pointers, performing zero table scans.
4. API Reference
5. Practical Examples
This example illustrates a FastAPI setup exposing an offset-based pagination endpoint with structured
response schemas.
app = FastAPI()
class ItemSchema(BaseModel):
id: int
name: str
class PaginatedResponse(BaseModel):
items: List[ItemSchema]
total_count: int
page: int
limit: int
# Mock Database
ITEMS_DB = [{"id": i, "name": f"Item {i}"} for i in range(1, 1000)]
@[Link]("/items", response_model=PaginatedResponse)
async def list_items(
page: int = Query(default=1, ge=1),
limit: int = Query(default=10, ge=1, le=100)
):
# Calculate skip offset
offset = (page - 1) * limit
total = len(ITEMS_DB)
paginated_items = ITEMS_DB[offset : offset + limit]
return {
"items": paginated_items,
"total_count": total,
"page": page,
"limit": limit
}
Always enforce strict upper bounds on pagination limits (e.g. le=100 inside the query parameter
declaration).
If you do not enforce limits, a client could request /items?limit=1000000 , forcing your web server to
allocate memory and load massive datasets, which can cause a crash.
7. Common Mistakes
Mistake: Calculating total_count using SELECT COUNT(*) on every paginated request on tables
containing millions of rows.
COUNT(*) is slow in PostgreSQL because it must perform a scan of the table to count rows. Running it
on every page request slows down performance.
Correction: Only run count queries on offset-based endpoints if mandatory for UI calculations. For
infinite scroll feeds (cursor-based), omit the total count entirely.
8. Performance Tips
When running cursor-based pagination, ensure that the columns used in your ORDER BY and cursor
comparison filters (like created_at or id ) have database indexes. If the order-by column is unindexed,
the database must sort the table in memory before applying the limit, slowing down queries.
9. Security Considerations
In cursor-based pagination, cursors are often returned to the client as base64-encoded strings
containing query metadata.
Mitigation: Sanitize and validate decoded cursor values (e.g., verifying that the decoded cursor is a
valid integer or date format) before passing them into SQL queries.
4. You will observe that the first item of Page 2 is a duplicate of the last item of Page 1. This illustrates
why cursor-based pagination is preferred for dynamic streams.
Cursor-based pagination is the standard architecture for feeds (like Twitter, Instagram, or chat message
logs) where users scroll down continuously. Because items are added constantly, using cursor
pagination ensures no duplicates are shown during scrolling.
Answer:
Offset-based pagination requires the database to scan, sort, and count all preceding rows before
returning the requested page. For large offsets (e.g. OFFSET 100000 ), this creates significant disk I/O
overhead.
Cursor-based pagination solves this by using a reference pointer (like WHERE id > 10000 ). The
database performs a quick index lookup to find that record and reads the next 10 rows, offering constant
O(1) execution time regardless of depth.
Question 2: Why can we skip COUNT(*) queries when using Cursor-based pagination?
Answer:
Cursor-based pagination is designed for dynamic lists (such as infinite scroll feeds) where users load
the next page sequentially rather than jumping to specific page numbers. Because the client only needs
to know if a next page exists (which can be checked by fetching limit + 1 rows), we can skip the
expensive table count query, improving performance.
13. Exercises
Write a function calculate_pagination_offset(page: int, limit: int) -> int that calculates the
database offset while validating boundaries.
Solution:
This mini-project is a product catalog API that implements cursor-based pagination. It encodes cursors
into base64 strings to abstract database implementation details from the client, validates inputs, and
checks if a next page exists.
import base64
from fastapi import FastAPI, Query, HTTPException
from pydantic import BaseModel
from typing import List, Optional
app = FastAPI()
class Product(BaseModel):
id: int
name: str
price: float
class PaginatedCursorResponse(BaseModel):
items: List[Product]
next_cursor: Optional[str]
class CursorSerializer:
@staticmethod
def encode(last_id: int) -> str:
"""Encodes database primary key into base64 string."""
return base64.b64encode(str(last_id).encode()).decode()
@staticmethod
def decode(cursor_str: str) -> int:
"""Decodes base64 cursor string to integer ID."""
try:
return int(base64.b64decode(cursor_str.encode()).decode())
except Exception:
raise HTTPException(status_code=400, detail="Invalid cursor token.")
@[Link]("/catalog", response_model=PaginatedCursorResponse)
async def get_catalog(
cursor: Optional[str] = Query(default=None, description="The base64 cursor
token"),
limit: int = Query(default=10, ge=1, le=20)
):
# Determine the starting ID
start_id = 0
if cursor:
start_id = [Link](cursor)
return {
"items": paginated_items,
"next_cursor": next_cursor_token
}
Key Takeaways
1. Pagination prevents web API crashes by splitting large database query results into manageable
pages.
2. Offset pagination ( LIMIT/OFFSET ) is simple and supports arbitrary page jumps, but degrades in
performance on deep pages and is prone to query drift.
3. Cursor pagination ( WHERE id > cursor ) offers constant-time O(1) search performance and
prevents drift, making it ideal for high-volume infinite scroll feeds.
4. Encrypt or base64-encode cursors to keep database column structures clean and abstracted from
public client integrations.
Further Reading
1. Introduction
Security is a core requirement for any web API. You must be able to:
2. Determine if that user has permission to perform the requested operation (Authorization).
In traditional monolithic applications, authentication was managed using Stateful Sessions. The server
created a session ID, stored it in a database or memory (like Redis), and sent a session cookie to the
client. On every request, the server read the cookie, queried the database, and retrieved the user's
session.
While reliable, stateful sessions do not scale well in modern distributed microservice architectures. If
you have dozens of isolated servers, they must all query a central session store to validate requests,
introducing a performance bottleneck.
Modern web services use Stateless Token Authentication, primarily using JSON Web Tokens (JWT).
In this chapter, we will learn how to hash passwords securely, analyze the structure of JSON Web
Tokens, and build a stateless JWT authentication system in FastAPI.
2. Theory
Never store raw, plain-text passwords in your database. If an attacker gains access to your database,
they will compromise all user accounts.
Hashing: Running a password through a one-way mathematical function (a cryptographic hash) that
produces a fixed-size string (the hash). It is mathematically impossible to reverse the hash back to
the password.
Salting: Adding a unique, random string of bytes (the salt) to the password before hashing it. This
ensures that if two users have the same password, they will produce completely different hashes,
protecting against Rainbow Table attacks (pre-compiled lists of common password hashes).
Slow Hashing Algorithms: Standard hashes like MD5, SHA-1, or SHA-256 are designed to be
fast, making them easy for attackers to brute-force using GPUs. Use slow, key-stretching algorithms
like Bcrypt or Argon2, which require significant CPU and memory to compute, making brute-force
attacks impractical.
A JSON Web Token (JWT) is a compact, URL-safe string containing three parts separated by dots ( . ):
[Link]
1. Header: A base64-encoded JSON block specifying the token type ( JWT ) and the signing algorithm
(e.g. HS256 ).
2. Payload: A base64-encoded JSON block containing claims (data about the user, like sub
(subject/user ID) and exp (expiration timestamp)).
Security Warning: The payload is base64-encoded, not encrypted. Anyone can decode a JWT
string and read the payload. Never store sensitive data (like passwords, keys, or SSNs) inside
the JWT payload.
3. Signature: Created by taking the encoded header and payload, joining them with a dot, and signing
them using a secret key and the algorithm specified in the header. The signature verifies that the
token has not been tampered with.
3. Internal Working
1. The server generates a JWT containing the user's ID and expiration time in the payload, signs it
using a secret key, and returns the JWT to the client.
2. The client stores the JWT and sends it in the Authorization header of subsequent requests:
Authorization: Bearer <token_string>
3. When a request arrives, the server extracts the JWT, decodes the signature, and verifies it using the
secret key.
If the signature matches, the server trusts the payload data (authenticating the user) without
needing to query a database or session store.
4. API Reference
# Configure Bcrypt
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# Hash a password
hashed = pwd_context.hash("user_secret_password")
# Verify password
is_valid = pwd_context.verify("user_secret_password", hashed)
Exposes the OAuth2 password bearer authentication scheme. It looks for an Authorization: Bearer
<token> header in requests and extracts the token.
This example uses the PyJWT library to sign and verify tokens.
import jwt
from datetime import datetime, timedelta
# Configuration keys
JWT_SECRET_KEY = "super-secret-key-change-in-production"
JWT_ALGORITHM = "HS256"
# Sign token
encoded_jwt = [Link](payload, JWT_SECRET_KEY, algorithm=JWT_ALGORITHM)
return encoded_jwt
If an attacker steals a user's JWT, they can use it to access your API until the token expires. Because
JWT authentication is stateless, you cannot easily revoke a token once it is issued.
Best Practice: Set a short expiration time on access tokens (e.g. 15 to 30 minutes). Use Refresh
Tokens to issue new access tokens without forcing the user to log in again.
7. Common Mistakes
Mistake: Using a simple string like "my_secret" as the JWT signing key.
Because JWT algorithms are public, an attacker can download your JWT and run brute-force tools (like
John the Ripper) to crack your secret key. Once cracked, they can forge valid tokens for any user.
Correction: Generate a strong cryptographic key and store it securely in environment variables:
8. Performance Tips
Verifying a JWT is purely computational (verifying the signature using CPU resources). It requires no
database lookup or network request, which makes JWT verification extremely fast and highly scalable.
9. Security Considerations
The header and payload of a JWT are only base64-encoded, not encrypted. Anyone can intercept the
token and read the user ID, email, or metadata inside the payload.
Mitigation: Never include passwords, bank details, roles configurations, or personal identifiable
information (PII) inside the JWT payload. Limit the payload to a simple identifier (such as user_id
or email ).
Copy the JWT string and paste it into the interactive debugger at [Link].
Inspect the decoded payload and header to verify that expiration timestamps ( exp ) are in the
correct Unix epoch format and that the fields contain the expected values.
In microservice architectures, an API Gateway authenticates users and issues a JWT. The downstream
microservices receive this JWT, verify the signature locally using a public key, and process requests
without needing to query a central database, ensuring high performance.
Question 1: What is the difference between stateful sessions and stateless JWT
authentication?
Answer:
Stateful Sessions require the server to store session data in memory or a database (like Redis)
and validate requests by checking the session ID against this store on every request. This
introduces a database query bottleneck.
Stateless JWT Authentication stores session data directly in the token payload on the client. The
server validates requests by verifying the cryptographic signature of the token using a secret key,
requiring no database queries.
Question 2: Why is it dangerous to store sensitive user data (like passwords or roles) in
a JWT payload?
Answer:
A JSON Web Token is base64-encoded, not encrypted. Anyone who intercepts the token can decode it
using standard base64 decoding tools and read the payload data in plain text. Storing sensitive data like
passwords or personally identifiable information (PII) inside a JWT payload exposes this data to theft
and violates security compliance rules.
13. Exercises
Write a helper class PasswordHelper containing two methods: hash_password(password: str) -> str
and verify_password(plain_password: str, hashed_password: str) -> bool using passlib .
Solution:
from [Link] import CryptContext
class PasswordHelper:
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
@classmethod
def hash_password(cls, password: str) -> str:
return cls.pwd_context.hash(password)
@classmethod
def verify_password(cls, plain_password: str, hashed_password: str) -> bool:
return cls.pwd_context.verify(plain_password, hashed_password)
This mini-project is a secure endpoint template. It implements password hashing, JWT token
generation, a dependency to extract and validate tokens, and an authenticated route.
# Mock User DB
MOCK_USERS_DB = {
"john_doe": {
"username": "john_doe",
"hashed_password": pwd_context.hash("secret123"), # Pre-hashed
"email": "john@[Link]"
}
}
# 2. Token Helpers
def generate_token(username: str) -> str:
payload = {
"sub": username,
"exp": [Link]() + timedelta(minutes=15)
}
return [Link](payload, SECRET_KEY, algorithm=ALGORITHM)
# 3. Authentication Dependency
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = [Link](token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = [Link]("sub")
if username is None:
raise credentials_exception
except [Link]:
raise credentials_exception
user = MOCK_USERS_DB.get(username)
if user is None:
raise credentials_exception
return user
@[Link]("/token")
async def login(form_data: Annotated[OAuth2PasswordRequestForm, Depends()]):
user = MOCK_USERS_DB.get(form_data.username)
if not user or not pwd_context.verify(form_data.password,
user["hashed_password"]):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password"
)
access_token = generate_token(user["username"])
return {"access_token": access_token, "token_type": "bearer"}
@[Link]("/users/me")
async def read_users_me(current_user: Annotated[dict, Depends(get_current_user)]):
return {"username": current_user["username"], "email": current_user["email"]}
Key Takeaways
2. Never store plain-text passwords. Use slow, GPU-resistant hashing algorithms like Bcrypt with
unique salts.
3. JSON Web Tokens (JWT) enable stateless authentication by encoding session data in the payload
and signing it with a secret key.
4. Access tokens must have short expiration times (e.g. 15-30 minutes). Secure documentation and
sensitive APIs using Bearer Token dependencies.
Further Reading
1. Introduction
Stateless JWT authentication has a major security limitation: once a JSON Web Token is issued, the
server cannot easily revoke it. If a user logs out, or if an attacker steals a token, the token remains valid
until it reaches its expiration time ( exp ).
If you set long expiration times (e.g. 7 days) to prevent forcing users to log in constantly, you expose
your application to high security risks. If you set short expiration times (e.g. 15 minutes), you improve
security, but users must enter their credentials every 15 minutes, resulting in a poor user experience.
1. Access Token: A short-lived (e.g., 15 minutes) stateless JWT used to authenticate API requests.
2. Refresh Token: A long-lived (e.g., 7 days) token stored securely on the client and registered in the
database on the server. It is used exclusively to request new access tokens.
In this chapter, we will design a secure Token Rotation system, explore the OAuth2 password flow, and
build a database-backed token refresh pipeline in FastAPI.
2. Theory
Access Token:
Refresh Token:
Storage: Sent in a secure, HTTP-only, SameSite cookie (protecting it from JavaScript access
and CSRF).
Verification: Stateful (verified against the database to support instant session revocation).
To detect and prevent token theft, production systems use Refresh Token Rotation.
Every time a client uses a refresh token to request a new access token, the server invalidates that
refresh token and issues a new one.
If an attacker steals a refresh token, they will attempt to use it. When the legitimate user also attempts
to use that same (now revoked or reused) token, the server detects the double-use attempt, flag it as a
breach, and invalidates all active sessions for that user, forcing them to log in again.
Legitimate User ---> Uses Refresh Token A ---> Server returns Access Token B &
Refresh Token C
Attacker ---> Tries Refresh Token A ---> Server detects reuse, revokes
all tokens!
3. Internal Working
1. Login: User supplies username/password. Server returns a short-lived Access Token and a long-
lived Refresh Token (stored in a database).
2. Accessing API: Client sends the Access Token in headers. If valid, the request succeeds.
3. Token Expiration: The Access Token expires. The client receives a 401 Unauthorized response.
4. Token Refresh: Client sends the Refresh Token to the /refresh endpoint.
If the refresh token exists and is valid: the server deletes the token, creates a new one, stores it
in the database, and returns a new Access Token and the new Refresh Token.
If the refresh token is missing or marked as revoked: the server suspects theft, revokes all active
refresh tokens for that user, and returns 400 Bad Request .
4. API Reference
To return a refresh token securely, send it inside an HTTP-only cookie using the Response object:
5. Practical Examples
This example demonstrates how to define a database model to track and revoke active refresh tokens
using SQLAlchemy.
class Base(DeclarativeBase):
pass
class UserToken(Base):
__tablename__ = "user_tokens"
Just like user passwords, never store refresh tokens in plain text in your database. If an attacker gains
access to your database tables, they can steal the refresh tokens and access user accounts.
Best Practice: Hash the refresh tokens (using SHA-256) before saving them to the database. When
verifying a token, hash the incoming client token and compare it with the stored hash.
7. Common Mistakes
8. Performance Tips
Ensure the token_hash column in your database has a Unique Index. Since the /refresh endpoint is
called frequently, looking up the token hash must run as a fast index scan to prevent database latency.
9. Security Considerations
If a user changes their password or reports a compromised account, execute a database query to
revoke all active refresh tokens associated with that user ID immediately:
This forces all active client sessions (browser, mobile apps) to log out instantly.
Verify that the cookie name matches, and check that the HttpOnly and Secure attributes are set
correctly.
Remember-Me Sessions
When users check a "Remember Me" box during login, the application issues a 30-day refresh token.
When they return to the website, the frontend uses the refresh token to silently obtain a new access
token, providing a seamless session experience without compromising security.
Question 1: Why split authentication into Access Tokens and Refresh Tokens?
Answer:
Access Tokens are stateless, short-lived JWTs. They provide fast authentication without querying
the database, but they cannot be revoked easily.
Refresh Tokens are long-lived and verified against the database. If an access token is
compromised, it expires quickly (e.g. 15 minutes). If a user logs out or a token is compromised, the
server revokes the refresh token in the database, blocking any future access requests.
Question 2: What is Refresh Token Rotation and how does it protect against token
theft?
Answer:
Refresh Token Rotation invalidates the current refresh token and issues a new one on every refresh
request. If an attacker steals a refresh token, they will attempt to use it. When the legitimate user also
attempts to use the same token, the database detects the double-use attempt, flags the breach, and
revokes all active tokens for that user, forcing a logout.
13. Exercises
Write a helper function hash_token(token: str) -> str that computes a secure SHA-256 hash of a
raw token string for database storage.
Solution:
import hashlib
This mini-project is a secure token rotation service. It implements password checks, token generation,
database token tracking, rotation validation, and a revocation endpoint.
import hashlib
import secrets
from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, status, Response, Cookie
from pydantic import BaseModel
from typing import Optional
class TokenService:
@staticmethod
def generate_random_token() -> str:
return secrets.token_urlsafe(32)
@staticmethod
def hash(token: str) -> str:
return hashlib.sha256([Link]()).hexdigest()
@classmethod
def register_refresh_token(cls, username: str, raw_token: str):
token_hash = [Link](raw_token)
expiry = [Link]() + timedelta(days=7)
TOKENS_DB[token_hash] = {
"user": username,
"expires": expiry,
"revoked": False
}
@[Link]("/login")
async def login(username: str, password: str, response: Response):
# Verify User
if USERS_DB.get(username) != password:
raise HTTPException(status_code=401, detail="Invalid credentials")
# Generate Tokens
access_token = f"access-for-{username}" # Mock access token (In production,
use JWT)
raw_refresh_token = TokenService.generate_random_token()
@[Link]("/refresh")
async def refresh_tokens(response: Response, refresh_token: Optional[str] =
Cookie(default=None)):
if not refresh_token:
raise HTTPException(status_code=401, detail="Refresh token cookie missing")
token_hash = [Link](refresh_token)
token_record = TOKENS_DB.get(token_hash)
# 1. Validation Checks
if not token_record:
raise HTTPException(status_code=401, detail="Invalid token")
Key Takeaways
1. Access tokens are short-lived, stateless JWTs used for API authorization. Refresh tokens are long-
lived, stateful keys used to request new access tokens.
2. Store refresh tokens securely in HTTP-only, Secure, SameSite cookies to protect them from XSS
and CSRF attacks.
3. Refresh Token Rotation invalidates used refresh tokens and issues new ones, preventing session
theft.
4. Hash refresh tokens (SHA-256) before storing them in your database. If a token is compromised,
update the database to revoke active sessions instantly.
Further Reading
1. Introduction
Authentication verifies who a user is. Once identified, you must determine what they are allowed to do.
This is called Authorization.
In simple applications, developers often hardcode role checks directly inside endpoints:
@[Link]("/admin/settings")
def view_settings(user: User = Depends(get_current_user)):
if [Link] != "admin":
raise HTTPException(status_code=403, detail="Forbidden")
This approach is fragile. If you need to add a new role (like "manager" or "editor" ) with access to
these settings, you have to modify every endpoint.
A clean architecture uses Role-Based Access Control (RBAC). You define permissions (scopes) for
specific actions, assign those permissions to roles, and associate users with those roles. The code
checks for permissions, not roles, making the system highly flexible.
In this chapter, we will learn how to design an RBAC schema, use OAuth2 scopes, and write reusable
permission dependencies in FastAPI.
2. Theory
Example: Only users with the "admin" role can delete orders.
Pros: Simple to understand and implement. Covers 90% of business use cases.
Attribute-Based Access Control (ABAC): Authorization is determined by attributes of the user, the
resource, and the environment.
Example: A manager can edit an order only if the order belongs to their department and it is
accessed during business hours.
3. Internal Working
FastAPI features native support for scopes using the Security class and SecurityScopes utility.
1. You declare a dependency using Security instead of Depends , specifying the required scopes.
2. FastAPI executes the dependency and injects a SecurityScopes object containing the scopes
required by the endpoint.
3. The dependency checks these scopes against the authenticated user's permissions, raising a 403
Forbidden error if any scopes are missing.
4. API Reference
Using SecurityScopes
def get_current_active_user(
security_scopes: SecurityScopes,
token: str = Depends(oauth2_scheme)
):
# Access required scopes via security_scopes.scopes
for scope in security_scopes.scopes:
...
5. Practical Examples
This example shows how to write a reusable permission checker class in FastAPI.
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from typing import Annotated
app = FastAPI()
class User(BaseModel):
username: str
role: str
permissions: list[str]
@[Link]("/items")
async def create_item(user: Annotated[User,
Depends(PermissionChecker(["write:items"]))]):
return {"message": "Access granted to write items"}
@[Link]("/items")
async def delete_item(user: Annotated[User,
Depends(PermissionChecker(["delete:items"]))]):
return {"message": "Access granted to delete items"}
Avoid referencing role names directly inside your controllers or service classes.
7. Common Mistakes
Mistake: Declaring permissions as static strings inside route handler code, making schema updates
difficult.
Correction: Maintain a centralized Permissions configuration module (such as an Enum or class) to
prevent typing mistakes:
8. Performance Tips
Checking user permissions inside endpoints requires loading user roles and join tables from your
database.
Best Practice: Cache active user permissions in a Redis store or request-scoped cache. This
avoids database overhead on repeated permission lookups.
9. Security Considerations
When using FastAPI's SecurityScopes , the required scopes are compiled and documentated inside the
OpenAPI definition file automatically.
Navigate to /docs to see the required scopes listed next to each lock icon in Swagger UI.
In enterprise B2B applications, customers define their own roles and permissions for their employees.
The system stores this custom mapping in tables. When a request is made, a dependency loads the
tenant's custom role definition to check access privileges.
Answer:
Role-Based Access Control (RBAC) determines access based on a user's assigned role (e.g.
Admin, Editor). It is static and simple.
Attribute-Based Access Control (ABAC) determines access dynamically using attributes
associated with the user, the target resource, and the environment (e.g. checking if the user owns
the resource and is accessing it during business hours).
Question 2: Why should authorization code check for permissions (scopes) instead of
roles?
Answer:
Checking for roles (e.g. [Link] == "admin" ) creates tight coupling. If you add a new role (like
"supervisor" ) with similar permissions, you have to modify every role check in the codebase.
Checking for permissions (e.g., write:orders ) keeps code decoupled. You can create roles and modify
permissions in the database without modifying your application code.
13. Exercises
Solution:
class RoleChecker:
def __init__(self, allowed_roles: list[str]):
self.allowed_roles = allowed_roles
This mini-project is a secure customer catalog API. It defines Users and Roles in a mock database,
implements a token parser with scope validations, and protects endpoints using OAuth2 specifications.
# Security Keys
SECRET_KEY = "rbac-super-secret-key"
ALGORITHM = "HS256"
oauth2_scheme = OAuth2PasswordBearer(
tokenUrl="token",
scopes={
"read:billing": "Read billing ledger details",
"write:billing": "Create or update billing transactions",
"admin:system": "System configuration access"
}
)
# 1. Custom User Schema
class AuthUser(BaseModel):
username: str
scopes: list[str]
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials signature",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# Decode token
payload = [Link](token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = [Link]("sub")
token_scopes: list[str] = [Link]("scopes", [])
if username is None:
raise credentials_exception
except [Link]:
raise credentials_exception
@[Link]("/billing")
async def get_billing(
# Secure endpoint using Security wrapper to enforce scopes
current_user: Annotated[AuthUser, Security(get_current_user, scopes=
["read:billing"])]
):
return {"message": "Billing data records", "authorized_user":
current_user.username}
@[Link]("/billing")
async def charge_billing(
current_user: Annotated[AuthUser, Security(get_current_user, scopes=
["write:billing"])]
):
return {"message": "Transaction charged", "authorized_user":
current_user.username}
Key Takeaways
4. Cache user roles and permission sets in high-performance datastores (like Redis) to avoid query
overhead.
Further Reading
1. Introduction
When users register for an account, you must verify that their email address is real and belongs to
them. If you do not verify emails, bot scripts can register fake accounts, clogging your database and
degrading email deliverability.
Similarly, if a user forgets their password, you must provide a secure recovery path. Both workflows:
If these tokens are poorly designed (e.g. they do not expire or can be reused), attackers can hijack
customer accounts.
In this chapter, we will design a secure user verification lifecycle, implement email and password
recovery pipelines, and enforce strict security boundaries in FastAPI.
2. Theory
The token payload contains the user ID and expiration time (e.g. {"user_id": 42, "exp":
1719480000} ). The token is signed using a secret key.
A random string (token) is generated, hashed, and stored in a database table along with the
user ID, expiration time, and an is_used flag.
Vulnerable API: Returns "Email not found" if the email is not registered. This allows attackers to
scan your API to discover which emails have accounts.
Secure API: Always returns a generic response, such as: "If the email exists, a password
reset link has been sent." This protects user privacy.
3. Internal Working
To prevent token reuse (which allows an attacker to reuse a stolen password reset token), the server
must invalidate the token immediately after it is validated:
Reset Request ---> Generate Random Token ---> Save Hash in DB (is_used=False) --->
Email link to User
User Clicks Link ---> Validate Token Hash ---> Set is_used=True in DB ---> Update
Password
User Clicks Link Again ---> Database checks is_used ---> Rejects request!
4. API Reference
Itsdangerous is a standard Python library used to sign data securely for untrusted environments.
# Configure Serializer
serializer = URLSafeTimedSerializer("my-secret-key")
5. Practical Examples
This example demonstrates how to sign and verify temporary tokens using itsdangerous with custom
timeouts.
SECRET_KEY = "verification-portal-secret-key"
serializer = URLSafeTimedSerializer(SECRET_KEY)
SALT_RESET = "password-reset-salt"
def generate_reset_token(email: str) -> str:
return [Link](email, salt=SALT_RESET)
Never make your signup endpoint block while waiting for an SMTP server to send a verification email.
Use FastAPI's BackgroundTasks or Celery to send emails in the background. This allows the signup
endpoint to respond instantly (in milliseconds), keeping the API fast and responsive.
7. Common Mistakes
Mistake: Verifying a password reset token but failing to invalidate it once the password is changed. If an
attacker intercepts the user's browser history, they can use the same link to update the password again.
Correction: Store token hashes in a database table and set is_used = True immediately upon
validation.
8. Performance Tips
9. Security Considerations
Always use cryptographically secure random number generators (such as Python's secrets module) to
generate reset tokens. Avoid using [Link]() , which produces predictable sequences, allowing
attackers to guess tokens.
Check that the salt parameter matches exactly during both the signing ( dumps ) and verification
( loads ) steps. itsdangerous uses the salt to isolate namespaces, and mismatches will raise
BadSignature errors.
1. User submits registration form. Account status is set to is_active = False in the database.
3. User clicks the link. The API validates the token and updates the database: is_active = True .
4. The user can now log in, protecting the system from bot registration.
Question 1: How do you prevent account enumeration attacks during password reset
requests?
Answer:
To prevent account enumeration, never return an error message (like "Email not registered" ) if a
user requests a password reset for an email that does not exist.
Instead, return a generic success message, such as: "If the email address exists in our system,
a password reset link has been sent." This prevents attackers from scanning your API to discover
active user accounts.
Question 2: Why should password reset tokens enforce single-use validation? How do
you implement this?
Answer:
Password reset tokens must be single-use to prevent hijackers from reusing a token if they intercept the
user's browser history or email.
This is implemented by storing the generated token hash in a database table with an is_used boolean
flag. When the user submits the reset request, the server queries the database, verifies that is_used is
False , performs the password update, and sets is_used = True in the same transaction.
13. Exercises
Create a helper function verify_stateful_token(db_tokens: dict, token: str) -> bool that verifies
a token exists, is not expired, and has not been used.
Solution:
This mini-project is a user verification and password reset service. It implements user registration with
inactive states, sends mock verification emails, manages single-use reset tokens, and updates user
passwords securely.
import secrets
from datetime import datetime, timedelta
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks, status
from pydantic import BaseModel, EmailStr
from typing import Optional
class UserRegister(BaseModel):
username: str
email: EmailStr
password: str
# 2. Token Generator
class TokenFactory:
@staticmethod
def create_token(email: str, token_type: str) -> str:
raw_token = secrets.token_urlsafe(32)
token_hash = hashlib_sha256(raw_token)
expiry = [Link]() + timedelta(hours=2)
TOKENS_DB[token_hash] = {
"email": email,
"type": token_type,
"expires": expiry,
"used": False
}
return raw_token
@[Link]("/register", status_code=201)
async def register(payload: UserRegister, background_tasks: BackgroundTasks):
if [Link] in USERS_DB:
raise HTTPException(status_code=400, detail="Email already registered")
@[Link]("/verify")
async def verify_account(token: str):
token_hash = hashlib_sha256(token)
record = TOKENS_DB.get(token_hash)
@[Link]("/forgot-password")
async def forgot_password(email: EmailStr, background_tasks: BackgroundTasks):
# Enforce Account Enumeration Protection: Always return success response
response_msg = {"status": "success", "message": "If the account exists, a reset
link has been queued."}
user = USERS_DB.get(email)
if not user:
return response_msg
return response_msg
Key Takeaways
1. Email verification helps prevent bot registration and ensures data quality. Password recovery
pipelines provide a secure way for users to regain access to their accounts.
2. Verification and password reset workflows rely on generating and validating secure, temporary
tokens.
3. Prevent account enumeration attacks by returning generic success messages for password reset
requests, hiding whether an email is registered.
4. Enforce single-use validation on reset tokens by tracking token state in the database, setting
is_used = True immediately upon validation to prevent token reuse.
Further Reading
1. Introduction
Almost every modern web application requires handling file uploads—whether it is users uploading
profile pictures, invoices, PDF documents, or CSV spreadsheets.
To receive files over the network, HTTP uses a special encoding type called multipart/form-data .
When a client uploads a file, the request body is divided into multiple parts, each separated by a
boundary string, containing headers and binary byte chunks.
In FastAPI, handling file uploads incorrectly can quickly crash your server. If you load a 2GB file upload
directly into Python memory as raw bytes, your server will run out of RAM, triggering an Out-Of-Memory
(OOM) crash that drops all active user requests.
In this chapter, we will learn how to handle file uploads safely using UploadFile and bytes
parameters, implement chunk-based streaming for large files, and enforce strict payload size
boundaries.
2. Theory
1. bytes :
How it works: The entire file contents are read directly into server RAM as a Python byte string.
Syntax: file: bytes = File(...)
Best Use Case: Small text files or metadata files (under 1MB).
Warning: Avoid using bytes for general uploads. If multiple users upload moderate files
concurrently, server memory will be exhausted.
2. UploadFile :
How it works: Starlette handles the file stream. If the file is small (under 1MB), it is kept in
memory. If it exceeds 1MB, Starlette automatically writes it to a temporary file on disk (using
Python's SpooledTemporaryFile ), protecting server RAM.
Best Use Case: General file uploads (images, PDFs, videos, zip archives).
3. Internal Working
Memory Buffering and Tempfile Storage
The UploadFile object exposes standard asynchronous methods (like await [Link]() , await
[Link]() , and await [Link]() ) that read from this underlying buffer or temporary file on
disk, keeping memory footprint low.
4. API Reference
await [Link](size) : Reads a specific number of bytes (or the whole file if size is omitted)
from the file stream.
await [Link](data) : Writes bytes data to the file stream.
await [Link]() : Closes the file handle and deletes the temporary file from the disk.
5. Practical Examples
This example demonstrates how to validate the file size and type of an uploaded file before saving it to
a local upload directory.
app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True) # Create upload directory
try:
with open(destination_path, "wb") as f:
while True:
# Read chunks of 64KB
chunk = await [Link](64 * 1024)
if not chunk:
break
size += len(chunk)
if size > MAX_FILE_SIZE:
# Clean up file on disk before raising error
[Link]()
destination_path.unlink()
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail="File size exceeds the 10MB limit."
)
[Link](chunk)
finally:
await [Link]() # Release temporary system files
7. Common Mistakes
# Antipattern: Loads the entire file into memory, bypassing Tempfile protections!
content = await [Link]()
Correction: Read and write the file in small, incremental chunks (e.g. 64KB or 1MB) as shown in the
practical example.
8. Performance Tips
If your API endpoints serve file downloads to users, avoid reading the file into memory to return it.
Instead, use Starlette's FileResponse or StreamingResponse , which streams the file from disk in
chunks directly to the client socket:
@[Link]("/download/{filename}")
async def download_file(filename: str):
file_path = Path("uploads") / filename
if not file_path.is_file():
raise HTTPException(status_code=404, detail="File not found")
# Streams file in chunks automatically
return FileResponse(path=file_path, filename=filename)
9. Security Considerations
Clients can upload files with malicious filenames containing directory traversal sequences (e.g.
../../etc/passwd ). If you save the file using this filename directly, you will overwrite critical system
files.
Mitigation: Never use [Link] directly to write paths. Sanitize the filename using secure
utility wrappers, or generate a unique random UUID filename instead:
import uuid
If requests are rejected with a 400 Bad Request before your endpoint code runs, check that the client is
sending the request with the correct headers:
The form field key name must match the parameter name declared in your route handler (e.g. file
in file: UploadFile = File(...) ).
Large CSV uploads can be parsed asynchronously in chunks, processing rows incrementally and
writing them to the database without loading the entire spreadsheet into memory.
Answer:
bytes loads the entire file contents directly into the server's RAM as a byte string, which can cause
Out-Of-Memory crashes for large files.
UploadFile streams the file. Small files are kept in memory, but files exceeding a threshold (default
1MB) are written to a temporary file on disk, keeping the memory footprint low. It also exposes
metadata like MIME type and filename.
Question 2: How do you prevent directory traversal attacks when saving uploaded
files?
Answer:
To prevent directory traversal attacks, never trust the client-supplied [Link] for directory paths.
If saved directly, paths containing ../ can overwrite system files. Sanitize filenames using safety
wrappers or generate a unique UUID filename on the server (e.g., f"{uuid.uuid4()}{extension}" )
before saving it to disk.
13. Exercises
Solution:
class MimeTypeValidator:
def __init__(self, allowed_types: list[str]):
self.allowed_types = allowed_types
This mini-project is a secure media upload manager. It accepts large file uploads, validates file sizes
dynamically, sanitizes filenames using UUIDs, and writes payloads to a local directory in chunks.
import uuid
from fastapi import FastAPI, UploadFile, File, HTTPException, status
from pathlib import Path
class MediaUploadManager:
@staticmethod
async def process_and_save_upload(file: UploadFile) -> str:
"""Saves file in 1MB chunks, validating size and generating a secure UUID
name."""
# 1. Generate secure name
file_suffix = Path([Link]).suffix
if file_suffix.lower() not in (".jpg", ".jpeg", ".png", ".mp4", ".pdf"):
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Unsupported file extension."
)
secure_name = f"{uuid.uuid4()}{file_suffix}"
target_path = UPLOAD_FOLDER / secure_name
bytes_written = 0
try:
with open(target_path, "wb") as buffer:
while True:
# Read in 1MB chunks
chunk = await [Link](1 * 1024 * 1024)
if not chunk:
break
bytes_written += len(chunk)
if bytes_written > MAX_MEDIA_SIZE:
# Clean up
[Link]()
target_path.unlink()
raise HTTPException(
status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
detail="Upload limits exceeded. Max size allowed is
20MB."
)
[Link](chunk)
finally:
await [Link]()
return secure_name
@[Link]("/media/upload")
async def upload_media(file: UploadFile = File(...)):
filename = await MediaUploadManager.process_and_save_upload(file)
return {
"status": "success",
"stored_filename": filename,
"location": f"/storage/media/{filename}"
}
Key Takeaways
2. Use UploadFile instead of bytes to handle file uploads. UploadFile streams files and uses
temporary files on disk for large files, protecting server memory.
3. Read large file uploads in incremental chunks (e.g. 64KB or 1MB) to keep memory usage low.
4. Prevent directory traversal attacks by generating unique random UUID filenames on the server
rather than using the client-supplied filename directly.
Further Reading
1. Introduction
In a production environment, you should never store user-uploaded files on the local disk of your web
server container. Modern applications run inside ephemeral containers (like Docker on Kubernetes or
AWS ECS) that are stateless and get destroyed and recreated during deployments or scaling events.
Any file saved to the local disk of a container is lost when the container restarts.
Furthermore, local disk storage does not scale horizontally. If you run 5 replicas of your FastAPI
application container behind a load balancer, a file uploaded to Container 1 is inaccessible to users
whose requests are routed to Container 2.
Production applications use Object Storage services like AWS S3 (Simple Storage Service), Google
Cloud Storage, or self-hosted alternatives like MinIO.
In this chapter, we will learn how to integrate FastAPI with AWS S3 and MinIO, upload files
asynchronously, and generate secure Presigned URLs to delegate file transfers directly to the cloud.
2. Theory
Key: The unique string identifier for the file (e.g. avatars/[Link] ).
Presigned URLs
Normally, your S3 bucket should be private to protect sensitive user files. If a user needs to download a
private file, your API can generate a Presigned URL.
A Presigned URL is a temporary URL generated by your server using your AWS credentials. It
embeds a cryptographic signature that grants access to the specific file for a short period (e.g. 15
minutes).
You can generate presigned URLs for GET (downloading files) and PUT/POST (allowing clients to
upload files directly to S3 without routing the file bytes through your FastAPI server, saving server
bandwidth and memory).
3. Internal Working
The official AWS SDK for Python ( boto3 ) is synchronous and blocks the event loop when making
network calls to S3. To perform async cloud operations, we use:
aioboto3 : An asynchronous wrapper around boto3 that uses aiohttp under the hood.
Alternatively, wrap synchronous boto3 calls inside asyncio.to_thread to run them in the worker
threadpool.
4. API Reference
Configuring S3 Clients
S3 Endpoint: When using AWS S3, the client automatically resolves endpoint URLs. When using
MinIO (for local development), you must configure endpoint_url pointing to your local container.
5. Practical Examples
This example demonstrates how to generate secure presigned URLs to delegate file uploads and
downloads to S3 using the standard boto3 client wrapped in thread executors.
import boto3
import asyncio
from fastapi import FastAPI, HTTPException
app = FastAPI()
BUCKET_NAME = "my-private-bucket"
@[Link]("/files/{file_key}/download")
async def get_download_link(file_key: str):
# Offload blocking boto3 network calls to worker threads
url = await asyncio.to_thread(get_presigned_download_url, file_key)
return {"download_url": url}
@[Link]("/files/{file_key}/upload")
async def get_upload_link(file_key: str):
url = await asyncio.to_thread(get_presigned_upload_url, file_key)
return {"upload_url": url}
Best Practice: In production, deploy your FastAPI application using IAM Roles (e.g. AWS IAM
Roles for Service Accounts in Kubernetes). The AWS SDK automatically retrieves temporary
security credentials from the container environment, eliminating credential management overhead.
7. Common Mistakes
Mistake: Calling s3_client.upload_fileobj() inside an async def route directly. This will block the
event loop, freezing the application while the file is uploaded.
Correction: Always wrap synchronous boto3 calls inside asyncio.to_thread or use the async-native
aioboto3 library.
Mistake: Setting bucket access policies to public to make file sharing easy. This exposes all uploaded
user files (which may contain invoices or private IDs) to search indexers.
Correction: Keep buckets private, and generate short-lived presigned URLs dynamically to share
specific files safely.
8. Performance Tips
Direct Client-to-S3 Uploads
If your users upload large files (like videos or high-resolution images), do not upload the files through
your FastAPI server. Every upload consumes FastAPI process CPU, memory, and network bandwidth.
Best Practice: Have the client request a presigned upload URL, and then upload the file directly
from the browser to the S3 bucket. This bypasses your FastAPI server entirely, keeping it fast and
responsive.
9. Security Considerations
Ensure your S3 bucket configuration enforces Server-Side Encryption (SSE) by default. This ensures
that even if the physical storage disks are compromised, your data remains encrypted.
If the client receives an HTTP 403 Forbidden error when trying to use a presigned URL:
Check that your server's system clock is synchronized. S3 signature validation is timestamp-based,
and clock drift on your web server will produce invalid signatures.
Ensure the HTTP method matches exactly (e.g., if you generated the URL for a PUT request, you
must send a PUT request, not a POST ).
CDN Integration
For public assets (like product images), integrate your S3 bucket with a CDN (Content Delivery
Network) like AWS CloudFront. The CDN caches images globally, reducing read traffic to your S3
bucket and speeding up asset delivery for users.
Question 1: Why should you avoid storing user-uploaded files on the local disk of your
web server in production?
Answer:
Modern production environments run applications in stateless, ephemeral containers. If the container
restarts or scales down, any file saved to its local disk is deleted. Additionally, local storage does not
scale horizontally. If you run multiple server containers, a file uploaded to Container 1 is inaccessible to
Container 2. You must use shared, persistent Object Storage (like AWS S3).
Question 2: What is a Presigned URL, and how does it improve API performance?
Answer:
A Presigned URL is a temporary URL containing cryptographic credentials generated by the server. It
allows clients to read (GET) or write (PUT) objects directly to/from S3.
It improves performance by allowing clients to upload or download files directly from S3, bypassing the
FastAPI server. This saves server CPU, memory, and network bandwidth, keeping the web API
lightweight.
13. Exercises
Solution:
import boto3
import asyncio
s3 = [Link]("s3")
This mini-project implements a complete asynchronous S3 Storage Manager. It manages file uploads,
generates presigned URLs, deletes files, and integrates with FastAPI's lifespan setup.
import boto3
import asyncio
from fastapi import FastAPI, Depends, UploadFile, File, HTTPException
from typing import Annotated
class S3StorageManager:
def __init__(self, bucket_name: str):
[Link] = bucket_name
[Link] = [Link]("s3")
async def get_download_url(self, key: str, expires_in: int = 3600) -> str:
"""Generates presigned download url asynchronously."""
return await asyncio.to_thread(
self.generate_download_url_sync,
key,
expires_in
)
# Non-blocking upload to S3
await storage.upload_file(file, file_key)
return {
"status": "uploaded",
"s3_key": file_key,
"access_url": download_url
}
Key Takeaways
1. Local container disks are ephemeral. Use Object Storage (S3/MinIO) to store user files in
production.
2. Standard AWS SDK boto3 calls are blocking. Wrap them in asyncio.to_thread or use aioboto3
to keep the event loop non-blocking.
3. Keep S3 buckets private by default to protect user files. Expose private files using short-lived,
cryptographically signed Presigned URLs.
4. Delegate heavy file transfers by having clients upload directly to S3 using presigned upload URLs.
This keeps your FastAPI server lightweight.
Further Reading
1. Introduction
When users upload images (like profile pictures or product photos), they often upload raw,
uncompressed files directly from their cameras. These files can be 10MB or larger, containing millions of
pixels and metadata.
High Bandwidth Costs: Serving a 10MB image to users loading a profile page consumes
excessive network bandwidth.
Slow Load Times: Mobile users on slow networks will experience lag, leading to a poor user
experience.
Privacy Risks: Camera photos embed EXIF metadata containing exact GPS coordinates, creation
timestamps, and camera models, exposing user location data.
Image Processing optimizes images before storing them. You resize them to standard dimensions,
strip metadata, and compress them using modern formats like WebP.
However, image manipulation is a CPU-bound operation. If you process images directly inside your
asynchronous FastAPI endpoint, you block the event loop, freezing the application for all users.
In this chapter, we will learn how to process images safely using Pillow, convert formats to WebP, and
run CPU-intensive tasks inside worker threads.
2. Theory
CPU-Bound Concurrency
As discussed in Chapter 2, Python's event loop executes async tasks concurrently on a single thread by
yielding during network or file I/O operations.
I/O-Bound: Waiting on external databases or sockets. Yields control, event loop continues.
CPU-Bound: Heavy calculations, encryption, or image decoding. Runs continuously on the CPU,
blocking the thread and the event loop.
To prevent CPU-bound image operations from freezing your web server, offload them using
asyncio.to_thread (which executes the function in a separate thread in Starlette's threadpool).
WebP is a modern image format developed by Google. It provides superior lossless and lossy
compression for web images:
WebP lossy images are typically 25% to 34% smaller than comparable JPEG images.
Converting user uploads to WebP saves disk space and speeds up asset load times.
3. Internal Working
When Python opens a compressed image file (like a 5MB JPEG), it must decompress it in memory to
process the pixels.
A 5MB file can expand into 500MB of raw pixel data in RAM when opened.
Attackers use this to execute Decompression Bomb attacks by uploading specially crafted tiny
files that expand into gigabytes of data, crashing the server process due to memory exhaustion.
Pillow protects against this by default. It checks the image dimensions before opening it and raises a
DecompressionBombError if it exceeds a safety threshold (default is 89,478,485 pixels). Never
disable this limit in production.
4. API Reference
[Link](size) : Modifies the image in-place to create a thumbnail, maintaining the aspect
ratio.
[Link](mode) : Converts the pixel format (e.g. converting a PNG with transparency from
RGBA to RGB before saving as JPEG).
[Link](fp, format) : Writes the image data to a file or file-like object (like [Link] ).
5. Practical Examples
This example demonstrates how to process an uploaded image asynchronously, resize it to a standard
avatar size, strip EXIF metadata, convert it to WebP, and return it.
import io
from PIL import Image, ImageOps
import asyncio
from fastapi import FastAPI, UploadFile, File, HTTPException
from [Link] import StreamingResponse
app = FastAPI()
@[Link]("/avatar/optimize")
async def optimize_avatar(file: UploadFile = File(...)):
if not file.content_type.startswith("image/"):
raise HTTPException(status_code=400, detail="Uploaded file is not an
image.")
try:
# Offload the CPU-bound processing to a worker thread
optimized_bytes = await asyncio.to_thread(process_image_sync, raw_bytes)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Image processing failed:
{e}")
Always strip EXIF metadata from uploaded images before storing them or serving them to other users.
This prevents exposing sensitive user location data (GPS coordinates) embedded in photos, ensuring
compliance with privacy regulations (like GDPR).
7. Common Mistakes
Mistake: Performing Pillow resize operations directly inside async def routes.
This blocks the event loop thread, preventing the server from handling other concurrent requests until
the image processing completes.
Correction: Always wrap Pillow execution blocks in asyncio.to_thread or delegate them to
background task queues.
8. Performance Tips
When saving images in WebP format, set quality=80 (lossy compression). This reduces file size
significantly (often by 80% compared to raw files) while maintaining high visual quality.
9. Security Considerations
Keep Pillow's default decompression bomb checks active. If you process user-uploaded images, never
set Image.MAX_IMAGE_PIXELS = None in production, as this exposes your application to memory
exhaustion attacks.
To verify that EXIF metadata was successfully stripped during image optimization:
Run the output image through metadata extraction utilities (like exifread or online EXIF viewers).
Verify that the geolocation, timestamp, and device fields are empty.
When users register and upload an avatar, the application processes the image in the background:
1. Resizes it to 256x256 (for thumbnail delivery) and 512x512 (for profile viewing).
2. Converts the files to WebP.
3. Uploads the optimized assets to S3 and updates the user's profile URL.
Question 1: Why is image processing considered CPU-bound, and how does it affect
FastAPI's event loop?
Answer:
Image processing (decoding, resizing, compressing pixels) requires intensive mathematical operations
executed by the CPU. Unlike network or file operations, it does not yield control.
If executed directly inside an async def route, it runs on the main event loop thread, blocking it. The
server cannot process other concurrent requests until the image operation finishes.
Question 2: How do you protect your server from "Decompression Bomb" attacks when
processing user-uploaded images?
Answer:
Decompression bombs are tiny compressed files that expand into gigabytes of raw pixel data in memory
when opened, exhausting server RAM.
Pillow protects against this by checking image dimensions before opening it. It raises a
DecompressionBombError if the dimensions exceed a safety threshold (default is 89,478,485 pixels).
Never disable this threshold in production.
13. Exercises
Write a function convert_png_to_webp(png_bytes: bytes) -> bytes that converts a PNG image byte
string to WebP format.
Solution:
import io
from PIL import Image
import io
import asyncio
from PIL import Image, ImageOps
from fastapi import FastAPI, UploadFile, File, HTTPException
from pathlib import Path
class AvatarOptimizer:
@staticmethod
def crop_and_optimize(image_bytes: bytes) -> bytes:
"""Crops image to a square aspect ratio, resizes to 256x256, and saves as
WebP."""
# Pillow dimension safety check is active by default
with [Link]([Link](image_bytes)) as img:
# 1. Auto-rotate and strip EXIF metadata
img = ImageOps.exif_transpose(img)
@[Link]("/avatar/process")
async def process_avatar(file: UploadFile = File(...)):
if not file.content_type.startswith("image/"):
raise HTTPException(status_code=400, detail="Invalid file type.")
Key Takeaways
1. Image processing is CPU-bound. Running it inside async route handlers blocks the event loop
thread, freezing the application.
2. Offload all Pillow operations to worker threads using asyncio.to_thread to keep your application
non-blocking.
3. Convert images to the WebP format to reduce file sizes (by 25-30% compared to JPEG) and speed
up asset delivery.
4. Always strip EXIF metadata from uploaded images before storing or serving them to protect user
privacy.
Further Reading
1. Introduction
Every time a user visits your homepage, your API might query a database, join multiple tables, and
format the result as JSON. If the data does not change often (like a product catalog list or a blog feed),
running the same database query repeatedly wastes CPU and disk I/O, slowing down your server.
To solve this, we use Caching. Caching is the practice of storing pre-calculated query results in a fast,
temporary storage layer.
Redis (Remote Dictionary Server) is the industry standard for in-memory caching. It is a key-value
database that stores all data in RAM, allowing for sub-millisecond read and write speeds.
In this chapter, we will learn how caching works, configure an asynchronous Redis client, implement the
Cache-Aside pattern, and use Redis for fast session and rate-limiting operations.
2. Theory
The most common caching strategy in web development is the Cache-Aside (Lazy Loading) pattern:
Request arrives ---> Check if key exists in Redis (Cache Hit) ---> Return cached
JSON
|
(Cache Miss)
v
Query Primary Database (Postgres) ---> Write result to Redis --->
Return data
1. When a request arrives, check if the data exists in Redis (the cache).
3. Cache Miss: If it does not exist, query the primary database (PostgreSQL), write the result into
Redis with an expiration time, and return the data to the client.
Time-To-Live (TTL)
Cached data must eventually expire. If you never clear the cache, your users will see outdated info
forever.
We set a Time-To-Live (TTL) on every cached key. This tells Redis to delete the key automatically after
a specified time (e.g. 5 minutes). Once deleted, the next request will trigger a cache miss, reloading
fresh data from the database.
3. Internal Working
Like PostgreSQL, synchronous Redis clients block the calling thread during socket reads. To maintain
high throughput in FastAPI, we use the async-native client provided by the redis library (formerly
known as aioredis ).
This client uses non-blocking sockets and connection pooling, executing operations asynchronously on
the event loop.
4. API Reference
Core Commands:
await [Link](key, value, ex=seconds) : Writes a key-value pair with an expiration time
(TTL).
5. Practical Examples
This example demonstrates how to implement a Cache-Aside pattern for a product catalog query.
import json
from fastapi import FastAPI, Depends
import [Link] as aioredis
from typing import Annotated
app = FastAPI()
return db_products
By default, if Redis runs out of memory, it will crash or reject new writes.
Best Practice: Set a memory limit (e.g. maxmemory 2gb ) in your [Link] and configure the
eviction policy to allkeys-lru (Least Recently Used). This tells Redis to automatically delete the
oldest, least-used keys when memory limits are reached, preventing crashes.
7. Common Mistakes
8. Performance Tips
Always use a shared ConnectionPool instance rather than opening and closing connections to Redis
on every HTTP request. Reusing TCP connections eliminates handshake overhead.
9. Security Considerations
By default, Redis does not require password authentication. If your Redis port ( 6379 ) is exposed to the
public internet, attackers can scan it, extract your cached tokens or user data, and even delete your
database.
To verify that keys are being written and check their remaining TTL:
View active keys: KEYS * (Note: Avoid using KEYS * in production on large datasets as it blocks the
thread; use SCAN instead).
View remaining TTL: TTL catalog:products (returns seconds remaining).
IP Rate Limiting
Redis's atomic INCR and EXPIRE operations make it the ideal backend for API rate limiters, allowing
you to track and block requests from specific client IPs with minimal latency.
Answer:
The Cache-Aside pattern is a caching strategy where the application handles both database queries
and cache synchronization:
Question 2: Why should you avoid using KEYS * in production Redis instances?
Answer:
Redis is a single-threaded system. The KEYS * command scans the entire keyspace database
sequentially to find matches, which blocks the execution thread. On large databases, this can freeze
Redis for several seconds, causing timeouts and performance drops for all concurrent application
queries. Use the non-blocking SCAN command instead.
13. Exercises
Write a function invalidate_cache_keys(redis: Redis, keys: list[str]) that deletes multiple cache
keys asynchronously to clear outdated data.
Solution:
This mini-project implements an asynchronous API rate-limiter middleware. It tracks client IP hits using
Redis, limits users to a maximum of 5 requests per minute, and handles client notifications.
import time
from fastapi import FastAPI, Request, Response, HTTPException, status
import [Link] as aioredis
app = FastAPI()
class RateLimitMiddleware:
def __init__(self, limit: int = 5, window_seconds: int = 60):
[Link] = limit
[Link] = window_seconds
@[Link]("http")
async def enforce_limiter(request: Request, call_next):
client_ip = [Link] if [Link] else "unknown"
# Check rate limit
is_allowed = await limiter.check_rate_limit(redis_client, client_ip)
if not is_allowed:
return Response(
content="Too Many Requests. Rate limit exceeded. Try again in a
minute.",
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
media_type="text/plain"
)
@[Link]("/resource")
async def access_resource():
return {"message": "Success! You are within your API rate limit limits."}
Key Takeaways
1. Caching stores frequently read, slowly changing data in memory (RAM), reducing query latency to
sub-millisecond levels.
2. The Cache-Aside pattern checks the cache first, loads from the database on a cache miss, and
updates the cache with an expiration time (TTL).
3. Set an eviction policy (like allkeys-lru ) and memory limits in [Link] to prevent memory
exhaustion crashes.
4. Secure your Redis instances by requiring password authentication and binding the service strictly to
local interfaces.
Further Reading
1. Introduction
In Chapter 12, we learned how to run tasks in the background using FastAPI's built-in BackgroundTasks
class. However, built-in background tasks run within the same Python process. For heavy workloads
(like video transcoding, bulk CSV parsing, or image processing), this approach falls short:
Resource Contention: CPU-bound tasks consume web server CPU resources, slowing down API
response times for all users.
Lack of Persistence: If the web server container crashes or restarts, all pending tasks are lost.
Scale Bottlenecks: You cannot scale background execution independently from your web
endpoints.
Celery decouples execution. Your FastAPI application acts as a client that publishes tasks to a
Message Broker (like RabbitMQ). Separate Celery Workers (running on different processes or
servers) consume tasks from the broker, execute them, and store results in a Result Backend (like
Redis).
In this chapter, we will learn how to configure Celery, run RabbitMQ brokers, manage worker pools, and
build reliable asynchronous task pipelines.
2. Theory
[FastAPI App (Client)] ---> (Publishes Task) ---> [RabbitMQ (Message Broker)]
|
(Pulls Task)
v
[Redis (Result Backend)] <--- (Writes Result) <--- [Celery Worker Processes]
1. Celery Client (FastAPI): Registers task signatures and schedules execution by sending task
parameters to the broker.
2. Message Broker (RabbitMQ): A message queue manager that receives task messages, routes
them to queues, and coordinates task delivery to active workers.
3. Celery Worker: Clean, isolated Python processes that poll the broker, retrieve tasks, and execute
the code.
4. Result Backend (Redis): A storage layer where workers save task results (success status, return
payloads, or failure tracebacks), allowing the client to poll task execution status.
Redis is an in-memory database. If the Redis server runs out of memory or restarts, pending task
messages can be lost.
3. Internal Working
1. Celery serializes the function name and arguments into a JSON payload.
3. A Celery worker pulls the message, decodes the payload, imports the task module, and executes
the function.
4. Once completed, the worker writes the return value or exception details to the Redis result backend
under the task's unique UUID.
5. The client uses the task UUID to query Redis and retrieve the execution status.
4. API Reference
Instantiation:
celery_app = Celery(
"tasks_engine",
broker="pyamqp://guest@localhost//", # RabbitMQ connection URL
backend="redis://localhost:6379/0" # Redis backend URL
)
Triggering Tasks:
5. Practical Examples
This example defines a Celery instance, configures an asynchronous task, and calls it from a FastAPI
endpoint.
# In celery_worker.py
from celery import Celery
import time
# In [Link]
from fastapi import FastAPI
from celery_worker import generate_pdf_report, celery_app
from [Link] import AsyncResult
app = FastAPI()
@[Link]("/reports/trigger", status_code=202)
async def trigger_report(name: str):
# Trigger task asynchronously
task = generate_pdf_report.delay(name, {"records": [1, 2, 3]})
@[Link]("/reports/status/{task_id}")
async def get_report_status(task_id: str):
# Query Redis result backend for status
task_result = AsyncResult(task_id, app=celery_app)
return {
"task_id": task_id,
"status": task_result.status, # e.g. PENDING, SUCCESS, FAILURE
"result": task_result.result if task_result.ready() else None
}
Celery must serialize task arguments into JSON to transmit them over the network.
Good: my_task.delay(user_id=42) (Pass primitive types like integers or strings, and reload the
database records inside the task).
7. Common Mistakes
Mistake: Configuring Celery to serialize payloads using Python's pickle library to allow passing
complex objects.
pickle is vulnerable to Remote Code Execution (RCE). If an attacker gains access to your RabbitMQ
broker, they can inject malicious serialized payloads that execute arbitrary shell commands inside your
worker processes.
Correction: Enforce strict JSON serialization:
celery_app.[Link](
task_serializer="json",
accept_content=["json"],
result_serializer="json"
)
8. Performance Tips
When starting a Celery worker, you define the concurrency pool size using the -c flag:
celery -A celery_worker worker -c 4
CPU-Bound Tasks: Set concurrency equal to the number of CPU cores. Too many worker
processes will cause CPU context-switching overhead.
I/O-Bound Tasks (e.g. scraping web pages): You can set concurrency higher (e.g. 20-50 workers)
or use the Eventlet/Gevent worker pools.
9. Security Considerations
Do not run RabbitMQ using default guest/guest credentials exposed to the public internet.
Create a dedicated user with strict permissions on specific virtual hosts ( vhosts ).
Deploy Flower, a real-time web-based monitoring tool for Celery. It allows you to:
To run:
E-commerce websites trigger invoice generation tasks asynchronously after checkout. Workers
generate PDFs in the background and upload them to S3, while the client receives a download link,
keeping the checkout flow fast and responsive.
Question 1: Why should you pass entity IDs (like user_id ) to Celery tasks instead of full
ORM objects?
Answer:
Full ORM objects (like SQLAlchemy models) contain active socket connections to the database and
session context states that cannot be serialized into JSON. Passing them as arguments will fail.
Additionally, database records can change between the time a task is scheduled and the time it
executes. Passing the entity ID and reloading the record from the database inside the task ensures the
worker always operates on the most up-to-date database state.
Question 2: What is the difference between RabbitMQ and Redis when used as a
message broker for Celery?
Answer:
Redis is an in-memory database. While fast, it does not support advanced routing and is subject to
data loss if the Redis server restarts or runs out of memory.
13. Exercises
Define a Celery task that attempts to fetch an API endpoint, and automatically retries on network failures
using exponential backoff (retrying after 2s, 4s, 8s, etc.).
Solution:
@celery_app.task(bind=True, max_retries=5)
def fetch_api_data(self, url: str):
try:
response = [Link](url)
response.raise_for_status()
return [Link]()
except Exception as exc:
# Calculate retry delay exponentially
retry_delay = 2 ** [Link]
raise [Link](exc=exc, countdown=retry_delay)
This mini-project implements a complete Celery processing suite. It configures the Celery worker class,
defines an image compression task with retry logic, and exposes FastAPI endpoints to schedule tasks
and poll execution progress.
# In app/worker_app.py
import time
from celery import Celery
celery_engine = Celery(
"media_worker",
broker="pyamqp://guest:guest@localhost:5672//",
backend="redis://localhost:6379/0"
)
celery_engine.[Link](
task_serializer="json",
accept_content=["json"],
result_serializer="json"
)
@celery_engine.task(bind=True, max_retries=3)
def compress_video_task(self, video_id: int, quality: str) -> dict:
"""Simulates a heavy, CPU-bound video processing task."""
print(f"[Worker] Fetching video ID {video_id} metadata...")
return {
"video_id": video_id,
"status": "completed",
"output_path": f"/storage/compressed/vid-{video_id}.mp4"
}
@[Link]("/transcode/{video_id}", status_code=202)
async def transcode_video(video_id: int):
# Schedule task asynchronously
task = compress_video_task.delay(video_id, quality="1080p")
return {"task_id": [Link], "status": "transcoding_queued"}
@[Link]("/transcode/status/{task_id}")
async def get_transcode_status(task_id: str):
task_result = AsyncResult(task_id, app=celery_engine)
response = {
"task_id": task_id,
"status": task_result.status
}
if task_result.status == "PROGRESS":
response["progress"] = task_result.[Link]("percentage")
elif task_result.status == "SUCCESS":
response["result"] = task_result.result
elif task_result.status == "FAILURE":
response["error"] = str(task_result.info)
return response
Key Takeaways
1. In-process background tasks share resource limits with the web server. Use Celery to offload heavy
workloads to separate worker processes.
2. RabbitMQ is the preferred production message broker due to its advanced routing and delivery
acknowledgement features.
3. Pass only primitive types (like database IDs) as task arguments to ensure clean JSON serialization.
4. Monitor task execution queues, execution times, and worker states using the Flower dashboard.
Further Reading
1. Introduction
Traditional HTTP connections are stateless and unidirectional. The client opens a connection, sends a
request, receives a response, and the connection is closed. If the server has new data (like a new chat
message or a live stock price update), it cannot push it to the client. The client must ask for it, leading to
inefficient polling patterns.
For real-time applications (such as chat apps, dashboard widgets, or gaming), we use WebSockets.
WebSockets provide a stateful, full-duplex, persistent connection over a single TCP socket. After an
initial handshake, the connection remains open, allowing both client and server to send messages to
each other at any time with minimal header overhead.
In this chapter, we will learn how WebSockets operate, manage connection lifecycles in FastAPI, and
scale real-time broadcasting across multiple servers using Redis Pub/Sub.
2. Theory
HTTP:
WebSockets:
Bidirectional: Both client and server can push messages at any time.
Client Server
| |
|--- HTTP Upgrade Handshake ------>|
|<-- Upgrade Accepted (101) -------|
| |
|========= Active Socket ==========| (Connection remains open)
|--- Send Message ---------------->|
|<-- Push Live Update -------------|
When a client connects to your FastAPI server via WebSockets, the persistent socket is held in the
memory of that specific server instance.
If you scale your application horizontally to run 3 instances behind a load balancer:
3. Internal Working
FastAPI manages WebSockets using the WebSocket class. The lifecycle follows a strict sequence:
1. Accept: The client requests a handshake. The server accepts it: await [Link]() .
2. Communication Loop: The server runs a loop, receiving client messages ( await
websocket.receive_text() ) and pushing updates ( await websocket.send_text() ).
3. Close: If the client closes the browser tab or loses internet, the socket connection breaks. This
raises a WebSocketDisconnect exception inside the loop, which the server must catch to clean up
connection references.
4. API Reference
await websocket.receive_json() : Reads and parses the next incoming JSON frame.
5. Practical Examples
This example implements a connection manager class that tracks active sockets and broadcasts
messages to all connected clients.
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List
app = FastAPI()
manager = ConnectionManager()
WebSockets connections can die silently due to network dropouts or firewall terminations, leaving
"ghost" connection sockets open in server memory.
Best Practice: Configure your ASGI server (Uvicorn) to send periodic ping frames to clients. If the
client does not respond with a pong frame within a timeout limit, the server closes the socket,
releasing memory:
uvicorn main:app --ws-ping-interval 20 --ws-ping-timeout 10
7. Common Mistakes
Mistake: Running a while True read loop without wrapping it in a try/except block catching
WebSocketDisconnect .
If the client disconnects, the next read operation raises a WebSocketDisconnect error. If uncaught, it
crashes the route handler thread, leaking connection references in memory.
Correction: Always wrap read loops in try/except WebSocketDisconnect: blocks to handle cleanup.
8. Performance Tips
While WebSockets support raw binary frames, avoid using WebSockets for large file transfers (like
uploading images). WebSockets are designed for low-latency message streaming.
Use HTTP multipart uploads for files, and use WebSockets to push status updates.
9. Security Considerations
Because WebSockets start as an HTTP request, they are vulnerable to Cross-Site WebSocket
Hijacking (CSWSH) (where a malicious website triggers a WebSocket connection to your API on behalf
of an authenticated user).
Mitigation: Verify the request Origin header during the handshake, and authenticate the client
using a short-lived access token passed as a query parameter (e.g. [Link] )
since custom headers are not supported by standard browser WebSocket APIs.
Enter your endpoint URL (e.g. [Link] ), click Connect, and send messages to
inspect full-duplex communication.
Finance dashboards connect to WebSockets feeds to receive real-time currency and stock price
fluctuations. The server receives feed updates from an external exchange socket and broadcasts them
to active dashboard connections, providing real-time data updates.
Question 1: How does a WebSocket connection start? What is the "Upgrade" header?
Answer:
A WebSocket connection begins as a standard HTTP request containing a Connection: Upgrade
header and a Upgrade: websocket header. This is the WebSocket Handshake. If the server supports
WebSockets, it responds with an HTTP status code 101 Switching Protocols , accepting the upgrade.
The connection transitions from a stateless HTTP request-response cycle to a persistent TCP socket.
Question 2: Why do you need a message broker (like Redis Pub/Sub) when scaling
WebSockets horizontally?
Answer:
When scaling horizontally, clients are distributed across different server instances. A client connected to
Server 1 cannot receive messages from a client connected to Server 2 because their sockets are held
in different server memories.
A message broker (like Redis Pub/Sub) acts as a bridge. When Server 1 receives a message, it
publishes it to a Redis channel. Server 2 subscribes to the channel and broadcasts the message to its
locally connected clients, ensuring cross-server communication.
13. Exercises
Write a WebSocket endpoint /ws/json that accepts incoming JSON frames containing {"message":
"..."} and echoes them back to the client as JSON.
Solution:
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
@[Link]("/ws/json")
async def ws_json_endpoint(websocket: WebSocket):
await [Link]()
try:
while True:
# Enforce JSON-only formats
data = await websocket.receive_json()
await websocket.send_json({"echo": [Link]("message")})
except WebSocketDisconnect:
await [Link]()
14. Mini Project: Scalable Real-time Chat Room with Redis Pub/Sub
This mini-project implements a scalable chat application. It manages local client connections and uses
Redis Pub/Sub to synchronize and broadcast chat messages across multiple application servers.
import asyncio
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
import [Link] as aioredis
from typing import List
class ChatConnectionManager:
def __init__(self):
self.local_connections: List[WebSocket] = []
# Instantiate Manager
manager = ChatConnectionManager()
@app.on_event("startup")
async def startup_event():
# Start the background Redis channel listener
redis_client = [Link](connection_pool=redis_pool)
[Link].redis_pub = redis_client
asyncio.create_task(redis_listener(redis_client))
@[Link]("/ws/chat/{username}")
async def chat_endpoint(websocket: WebSocket, username: str):
await [Link](websocket)
redis_pub: [Link] = [Link].redis_pub
Key Takeaways
1. WebSockets provide persistent, full-duplex, bidirectional communication over a single TCP socket
connection.
2. WebSockets begin as an HTTP request containing an Upgrade header, which the server upgrades
to a permanent socket connection (status 101 ).
3. Always wrap your WebSocket read loops in try/except WebSocketDisconnect: blocks to close
sockets and release memory safely.
4. Scale WebSockets across multiple servers using a message broker (like Redis Pub/Sub) to
broadcast messages globally.
Further Reading
1. Introduction
Building a web API is more than just writing code that returns JSON. You are designing an interface that
other developers (and your own frontend team) will interact with for years.
A poorly designed API—one that lacks clear naming conventions, mixes versioning strategies, returns
inconsistent error schemas, or changes endpoint names randomly—degrades developer productivity
and leads to bugs.
2. Theory
Nouns, Not Verbs: Endpoints should represent resources (nouns), never actions (verbs).
Good: GET /users (list users) and POST /users (create a user)
GET : Retrieve a resource. Safe and idempotent (does not modify database state).
As your business grows, you will need to modify your API structure. To prevent breaking existing client
integrations, you must version your API.
Highly readable;
Pollutes URL
URL Path /api/v1/users easy to cache and
namespaces.
route.
Follows REST
Accept Accept: Complex to route
application/[Link].v1+json content-negotiation
Header and manage.
standards.
Best Practice: Use URL Path Versioning for public Web APIs. It is the most common approach
and is highly compatible with API gateways and cache proxies.
3. Internal Working
{
"status": 422,
"type": "/errors/validation-failed",
"title": "Validation Error",
"detail": "The request body failed schema validation checks.",
"errors": [
{
"loc": ["body", "email"],
"msg": "value is not a valid email address",
"type": "value_error.email"
}
]
}
4. API Reference
app = FastAPI()
# 1. Version 1 Router
v1_router = APIRouter(prefix="/api/v1")
@v1_router.get("/users")
def get_v1_users():
return [{"version": "v1"}]
# Register Routers
app.include_router(v1_router)
app.include_router(v2_router)
5. Practical Examples
This example overrides FastAPI's default exception handlers to return consistent RFC-7807 style error
responses.
app = FastAPI()
Once an API version (e.g. /api/v1 ) is deployed to production, never delete fields, change data types,
or modify endpoint paths.
If you need to make a breaking change (such as renaming first_name to firstName or dropping a
field), create a new version path ( /api/v2 ) and deprecate v1 gradually.
7. Common Mistakes
Mistake: Returning raw Python tracebacks or database errors to the client on server crashes. This
exposes server paths, database credentials, or column names to attackers.
Correction: Catch raw exceptions in a global error handler and return a generic error payload (like the
500 error handler in the practical example) to protect your systems.
8. Performance Tips
If your database contains tables with many columns, allow clients to request only the specific fields they
need:
GET /api/v1/users?fields=id,username
This reduces database query payload sizes and saves network bandwidth, speeding up load times.
9. Security Considerations
1. Sanitize Sorting Fields
If you allow clients to define database sorting parameters dynamically using query inputs (e.g. GET
/api/v1/items?sort=price ):
Mitigation: Validate the sort parameter against a strict whitelist of allowed columns (e.g. ["price",
"created_at"] ) before passing it to your SQL query builder. Passing arbitrary query strings directly
to database sorting clauses exposes your application to SQL Injection.
To verify that your routers are prefixing versions correctly, write a startup utility that prints all active
routes:
@app.on_event("startup")
def print_routes():
for route in [Link]:
print(f"Path: {[Link]} | Methods: {[Link]}")
Companies like Stripe or Twilio maintain legacy API versions for years to ensure that third-party client
integrations do not break when system schemas are updated.
Question 1: What is the difference between path versioning and accept header
versioning?
Answer:
URL Path Versioning includes the version directly in the path (e.g. /api/v1/users ). It is highly
readable and easy to route and cache, but it can clutter URL namespaces.
Accept Header Versioning uses content negotiation, sending the version in the headers (e.g.,
Accept: application/[Link].v1+json ). It keeps URLs clean, but is more complex to test and
route.
Answer:
RESTful design focuses on Resources (entities like Users, Orders, or Products). Using nouns (e.g.
/users ) represents the resource collection, while HTTP methods (GET, POST, DELETE) specify the
actions to perform. This makes APIs clean, intuitive, and standard.
13. Exercises
Write a dependency get_sort_order(sort: str = "id") that validates a sort query string, raising a
400 Bad Request if it is not in the whitelist ["id", "price", "created_at"] .
Solution:
This mini-project is a production-grade REST API scaffolding module. It configures v1 and v2 routes
using APIRouter , implements a global error-handling boundary that maps Python exceptions to
standardized JSON schemas, and supports request filters.
class ProductV1(BaseModel):
id: int
name: str
price: float
MOCK_PRODUCTS = [
{"id": 1, "name": "Desk Chair", "price": 150.00, "category": "furniture"},
{"id": 2, "name": "Desk Lamp", "price": 45.00, "category": "lighting"}
]
@v1_router.get("/products", response_model=List[ProductV1])
async def list_products(
category: Optional[str] = Query(default=None),
min_price: Optional[float] = Query(default=None, ge=0)
):
"""Endpoint supporting V1 query filters."""
filtered = MOCK_PRODUCTS
if category:
filtered = [p for p in filtered if p["category"] == category]
if min_price is not None:
filtered = [p for p in filtered if p["price"] >= min_price]
return filtered
@v1_router.get("/products/{product_id}", response_model=ProductV1)
async def get_product(product_id: int):
product = next((p for p in MOCK_PRODUCTS if p["id"] == product_id), None)
if not product:
# Raises exception; formatted dynamically by exception boundary handler
raise HTTPException(status_code=404, detail="Product not found.")
return product
@v2_router.get("/products", response_model=List[ProductV2])
async def list_products_v2():
# Maps internal structure to the new V2 schema
return MOCK_PRODUCTS
Key Takeaways
1. RESTful design centers on resources (nouns) manipulated using standard HTTP methods (GET,
POST, PATCH, DELETE).
2. Use URL path versioning (e.g. /api/v1/ ) to manage schema evolution without breaking existing
integrations.
4. Implement global exception handlers to map raw Python exceptions to standardized RFC 7807
JSON error responses, protecting server internals from data leaks.
Further Reading
1. Introduction
When you expose a web API to the public internet, it will be targeted by automated vulnerability
scanners and malicious actors. If you do not secure your endpoints, attackers can exploit flaws to
extract user data, hijack sessions, bypass billing gates, or take down your servers.
Securing an API requires applying defense-in-depth principles at every layer of your application.
In this chapter, we will analyze the OWASP API Security Top 10 vulnerabilities (including BOLA/IDOR,
Mass Assignment, and SSRF), learn how to configure Cross-Origin Resource Sharing (CORS) securely,
enforce HTTP security headers, and build strict authorization checks in FastAPI.
2. Theory
The Open Web Application Security Project (OWASP) maintains a list of the most critical security risks
for APIs. Here are the core issues and how to defend against them:
The Risk: A user accesses another user's private data by simply changing the ID in the request
URL: GET /api/v1/orders/99 (where order 99 belongs to someone else).
The Defense: Never trust that a user is allowed to access an object simply because they know
its ID. Always verify that the authenticated user owns or is authorized to access the requested
resource.
2. API3: Broken Object Property Level Authorization (Mass Assignment / Excessive Data
Exposure):
The Risk:
Excessive Data Exposure: The API queries the database and returns the raw user row—
including password hashes and verification tokens—relying on the frontend to hide these
fields in the UI.
Mass Assignment: The API maps request payloads directly to database columns. An
attacker submits a POST request containing "is_admin": true or "role": "admin" ,
updating their privileges.
The Defense: Always use Pydantic schemas to explicitly define allowed input ( request model )
and output ( response model ) fields. Never read or write database attributes directly from
unvalidated client inputs.
The Risk: An endpoint accepts an external URL parameter, fetches data from that URL, and
returns it. An attacker submits a local URL (e.g. [Link] ), tricking the
server into calling local, restricted APIs behind the firewall.
The Defense: Avoid allowing user-supplied URLs to be fetched directly by the server. If
required, validate URLs against a strict whitelist of domains, and isolate the outbound HTTP
client in a restricted network zone.
3. Internal Working
Cross-Origin Resource Sharing (CORS)
CORS is a browser security mechanism that restricts web pages from making requests to a different
domain than the one that served the page.
|
[Blocked if
not allowed!]
The Preflight Request: For state-modifying requests (like POST or PATCH), the browser sends an
options request ( OPTIONS ) first, asking the server if the origin is allowed.
4. API Reference
app = FastAPI()
5. Practical Examples
This example demonstrates how to check resource ownership inside a FastAPI dependency.
from fastapi import FastAPI, Depends, HTTPException, status
from pydantic import BaseModel
from typing import Annotated
app = FastAPI()
class User(BaseModel):
id: int
username: str
class Order(BaseModel):
id: int
owner_id: int
total: float
# Mock Database
ORDERS_DB = {
101: Order(id=101, owner_id=1, total=99.99),
102: Order(id=102, owner_id=2, total=19.99) # Belongs to a different user
}
Add security headers to every response to protect clients from Cross-Site Scripting (XSS), Clickjacking,
and MIME sniffing attacks:
Content-Security-Policy (CSP) : Restricts where scripts and resources can be loaded from.
7. Common Mistakes
8. Performance Tips
CORS preflight ( OPTIONS ) requests add network latency before every actual request.
Optimization: Set the max_age parameter in your CORSMiddleware configuration (e.g. max_age=600
seconds). This tells browsers to cache the preflight response, reducing network round-trips for
subsequent requests.
9. Security Considerations
Inspect the output to verify that X-Frame-Options , Content-Security-Policy , and other security
headers are present.
Financial endpoints (like checking bank balances or charging cards) implement strict BOLA checks,
enforce TLS client certificates, and restrict IP origins to protect customer transactions.
Answer:
BOLA (Broken Object Level Authorization), also known as IDOR (Insecure Direct Object Reference),
occurs when an API allows a user to access a resource by simply guessing or modifying its ID in the
request URL.
To prevent BOLA, never trust that a user is allowed to access an object simply because they know its
ID. Always verify that the authenticated user's ID matches the owner ID of the requested resource inside
your validation dependencies before returning data.
Question 2: Why should you avoid wildcard * origins in CORS settings for
authenticated APIs?
Answer:
If you configure CORS origins to allow wildcards ( * ) while enabling credentials, you allow any website
(including malicious domains) to execute JavaScript code that sends authenticated requests to your API
on behalf of your users. This allows attackers to steal user data and execute unauthorized actions.
13. Exercises
Solution:
class UserUpdate(BaseModel):
# Only allow safe columns to be passed during updates
username: str = Field(..., max_length=50)
bio: str = Field(..., max_length=250)
This mini-project is a security hardening module. It implements strict CORS validations, secure security
headers (HSTS, Content-Security-Policy, X-Frame-Options), an ownership check dependency to
prevent BOLA/IDOR attacks, and payload input validation limits.
return response
# Mock Models
class Account(BaseModel):
id: int
user_id: int
balance: float
MOCK_ACCOUNTS = {
10: Account(id=10, user_id=1, balance=5000.00),
20: Account(id=20, user_id=2, balance=25.00)
}
# Verify ownership
if account.user_id != self.user_id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Access Denied. Resource ownership validation failed."
)
return account
Key Takeaways
1. Broken Object Level Authorization (BOLA/IDOR) is the most common API security risk. Always
verify resource ownership before returning data.
2. Defend against Mass Assignment and Excessive Data Exposure by using Pydantic schemas to
validate and filter all API inputs and outputs.
3. Explicitly define CORS origins instead of using wildcards ( * ) to prevent malicious sites from
accessing credentials.
Further Reading
1. Introduction
You should never deploy an application to production without automated tests. Manual testing (clicking
through Swagger UI or Postman) is slow, prone to human error, and does not scale.
Automated testing verifies that your application behaves correctly when you modify code, refactor
database schemas, or upgrade third-party packages.
You must configure tests to run asynchronously using an async test client ( AsyncClient ).
You must mock external services (like payment gateways or email services) so tests do not run real
operations during execution.
You must isolate database operations so that tests do not write garbage data into your production
database or conflict with each other when run concurrently.
In this chapter, we will learn how to configure Pytest, write async integration tests, override
dependencies, and isolate database tests using transactional rollbacks.
2. Theory
The Testing Pyramid
1. Unit Tests: Test individual functions or isolated classes in isolation. They are fast and run without
databases or network connections.
2. Integration Tests: Test how different components work together. In FastAPI, this means sending
request payloads to endpoints and verifying database updates and response outputs.
3. End-to-End (E2E) Tests: Test the entire application flow from the user's perspective, typically
simulating a browser interacting with the frontend and backend.
3. Internal Working
To mock external integrations (like a billing service or third-party login APIs) during testing, FastAPI
allows you to temporarily override dependencies registered on your application:
# During testing:
app.dependency_overrides[get_payment_client] = mock_payment_client
When a test client sends a request to an endpoint that requires get_payment_client , FastAPI executes
your mock_payment_client function instead, preventing real billing charges from running during tests.
4. API Reference
FastAPI's built-in TestClient is synchronous. For async route endpoints, use HTTPX's AsyncClient :
@[Link]
async def test_endpoint():
async with AsyncClient(app=app, base_url="[Link] as ac:
response = await [Link]("/health")
assert response.status_code == 200
5. Practical Examples
This example implements a test suite using pytest . It configures database isolation fixtures and
overrides dependencies to mock external network requests.
# In test_main.py
import pytest
from httpx import AsyncClient
from fastapi import FastAPI, Depends
from [Link] import create_async_engine, async_sessionmaker,
AsyncSession
from typing import Annotated
app = FastAPI()
@[Link]("/checkout")
async def checkout(status: Annotated[str, Depends(check_payment_status)]):
return {"message": f"Order processed. Status: {status}"}
@[Link]
def anyio_backend():
return "asyncio"
@[Link]
async def test_checkout_endpoint():
# 2. Register Dependency Override
app.dependency_overrides[check_payment_status] = mock_payment_status
# 4. Clean up override
app.dependency_overrides.clear()
Always clean up dependency overrides after your tests run. If you forget to clear the overrides
( app.dependency_overrides.clear() ), subsequent tests will continue to use the mocked dependencies,
leading to false positives or test conflicts.
7. Common Mistakes
Mistake: Running tests against a local development database without resetting state, causing tests to
fail randomly depending on the order they are run.
Correction: Create a dedicated test database (e.g. test_db ) and use transactional fixtures that roll
back writes after every test.
Mistake: Decorating async tests with @[Link] but forgetting to define the anyio_backend
fixture. This raises configuration errors because Pytest does not know which event loop implementation
(asyncio or trio) to use.
Correction: Add the anyio_backend fixture returning "asyncio" in your [Link] file.
8. Performance Tips
If your unit tests do not rely on PostgreSQL-specific features (like JSONB or full-text search), use an in-
memory SQLite database ( sqlite+aiosqlite:///:memory: ) for your test suite. SQLite running in RAM
is extremely fast, allowing thousands of tests to run in seconds.
9. Security Considerations
Ensure your test runner is blocked from executing against your production database.
Mitigation: Add safety checks in your [Link] configuration to verify that the database URL
does not contain production hostnames before creating engine connections.
By default, Pytest captures and hides standard outputs during execution. If your debug print statements
do not appear in the terminal:
CI pipelines run your automated test suite on every pull request. If a developer introduces a bug that
breaks user registration or order checkout, the pipeline fails, blocking the code from being merged or
deployed.
Answer:
FastAPI provides a global overrides map: app.dependency_overrides . You register a mock dependency
by mapping the original dependency function to the mock function:
app.dependency_overrides[original_dependency] = mock_dependency
FastAPI will execute the mock function instead of the original one. Clear overrides after the test using
app.dependency_overrides.clear() .
Answer:
The database rollback pattern runs each test inside an isolated database transaction. When the test
completes, the transaction is rolled back: await [Link]() .
This ensures that any data written during the test is discarded. The database remains clean, preventing
conflicts between tests, and avoiding the slow overhead of recreating tables for every test.
13. Exercises
Write an async test using AsyncClient that verifies a GET request to /health returns status code 200
and payload {"status": "ok"} .
Solution:
import pytest
from httpx import AsyncClient
@[Link]
async def test_health_check(app):
async with AsyncClient(app=app, base_url="[Link] as ac:
res = await [Link]("/health")
assert res.status_code == 200
assert [Link]() == {"status": "ok"}
This mini-project implements a complete test configuration pipeline inside a [Link] file. It sets up
an async in-memory SQLite database, wraps each test in an isolated database transaction block,
overrides the database session dependency, and executes a full CRUD endpoint integration test.
# In [Link]
import pytest
from [Link] import create_async_engine, async_sessionmaker,
AsyncSession
from [Link] import DeclarativeBase, Mapped, mapped_column
from fastapi import FastAPI, Depends
from typing import Annotated
class Base(DeclarativeBase):
pass
class DBProduct(Base):
__tablename__ = "products"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
@[Link](scope="session", autouse=True)
async def setup_test_database():
"""Compiles tables in the test database once per test run."""
async with [Link]() as conn:
await conn.run_sync([Link].create_all)
yield
async with [Link]() as conn:
await conn.run_sync([Link].drop_all)
await [Link]()
@[Link]
async def db_session() -> AsyncSession:
"""Fixture to execute each test within an isolated database transaction."""
async with [Link]() as connection:
# 1. Begin transaction block
transaction = await [Link]()
# Bind session to the transaction connection
session = TestingSessionLocal(bind=connection)
yield session
@[Link]("/products", status_code=201)
async def create_product(name: str, db: ActiveSession):
product = DBProduct(name=name)
[Link](product)
await [Link]()
return {"id": [Link], "name": [Link]}
# Clear overrides
app.dependency_overrides.clear()
Key Takeaways
1. Use automated unit, integration, and E2E tests to verify application correctness before deploying
updates to production.
2. FastAPI's synchronous TestClient is unsuitable for async endpoints. Use HTTPX's AsyncClient
instead.
3. Mock external integrations (like APIs) during tests using FastAPI's app.dependency_overrides map.
4. Isolate database tests using the transactional rollback pattern to prevent tests from writing garbage
data or conflicting with each other.
Further Reading
1. Introduction
Deploying raw Python applications directly to virtual machines creates dependency challenges:
The production server might run a different version of Python than your local machine.
Installing system packages (like compilation headers or database drivers) can cause conflicts with
other services on the server.
Managing application upgrades cleanly is difficult, leading to runtime differences ("it works on my
machine").
Containerization solves this by packaging your application code, runtime dependencies, libraries, and
configuration files into an isolated container image.
Docker is the industry standard for containerization. A container runs as an isolated process on the host
OS kernel, providing a consistent environment from development to production.
In this chapter, we will learn how to write optimized, secure, multi-stage Dockerfiles for FastAPI,
configure cache-friendly builds, and orchestrate multi-container environments using Docker Compose.
2. Theory
Multi-Stage Builds
In a naive Docker build, compiler tools and temporary cache files are left inside the final image, resulting
in large file sizes (often over 1GB) and increasing the security attack surface.
Multi-Stage Builds solve this by using multiple FROM instructions in a single Dockerfile.
Stage 1 (Builder): Installs build-essential tools (compilers, git) and compiles Python
dependencies into a virtual environment or wheels.
Stage 2 (Runner): Starts from a clean, minimal base image (like python-slim ), copies only the
compiled virtual environment and code, and discards all build tools, reducing the final image size
(often under 200MB) and keeping it secure.
+------------------------------------+
| Builder Stage: |
| - Load full compiler tools | ---> Compiles dependencies inside venv
| - Pip install requirements |
+------------------------------------+
| (Only copy venv & code)
v
+------------------------------------+
| Runner Stage: |
| - Load minimal Python-slim image | ---> Final lightweight production image
| - Runs as non-root user |
+------------------------------------+
Docker Layer Caching
Docker builds images sequentially. Each command in a Dockerfile creates a read-only layer.
If a layer has not changed, Docker reuses the cached layer from previous builds, speeding up
compile times.
If a layer changes, all subsequent layers are invalidated and must be rebuilt.
Best Practice: Copy only [Link] and install dependencies before copying your
application source code. Since code changes frequently but dependencies change rarely, this allows
Docker to cache the slow dependency installation step.
3. Internal Working
By default, Docker container processes run as the root user. If an attacker exploits a code vulnerability
(like directory traversal or remote code execution) inside a root-running container, they can escape the
container isolation barrier and gain administrative control over the host operating system.
Always define and switch to a dedicated non-privileged user (e.g. appuser ) in your runner stage
Dockerfile before launching the application.
4. API Reference
5. Practical Examples
This Dockerfile compiles a virtual environment in a builder stage and runs the FastAPI application as a
non-root user in a minimal slim runner stage.
# =========================================================
# STAGE 1: Builder
# =========================================================
FROM python:3.11-buster AS builder
WORKDIR /build
# =========================================================
# STAGE 2: Runner (Lightweight & Secure)
# =========================================================
FROM python:3.11-slim-buster AS runner
WORKDIR /app
To prevent copying local files like .venv , test reports, or Git folders into the image, create a
.dockerignore file:
# .dockerignore
.git
.github
.venv
__pycache__
*.pyc
tests
Dockerfile
[Link]
Never use the latest tag in your base images (e.g. FROM python:latest ).
The latest tag references whatever version was pushed last. If the base image is updated (e.g.
from Python 3.11 to 3.12), your next build could fail due to incompatibilities. Always use explicit tags
(e.g. FROM python:3.11-slim ).
7. Common Mistakes
# Antipattern
COPY . .
RUN pip install -r [Link] # Runs on every tiny code change!
Correction: Copy only [Link] , run pip install , and then copy the remaining source code
files.
8. Performance Tips
Use --no-cache-dir in Pip
When installing python dependencies, always add the --no-cache-dir flag to pip install . This
prevents pip from storing downloaded wheel archives inside the image layer, saving disk space.
9. Security Considerations
Regularly pull base image updates and rebuild your container images. This ensures that any operating
system package security vulnerabilities (CVEs) inside the base image layers are patched, keeping your
production deployments secure.
If a container fails to start or behaves unexpectedly, inspect it dynamically by running an interactive shell
session inside the running container:
docker exec -it [container_id] /bin/bash
This allows you to verify configuration paths, view permissions, or check environment variables
manually.
Modern cloud deployments push compiled Docker images to registries (like AWS ECR or Docker Hub)
and trigger rollouts on Kubernetes clusters, providing scalable and repeatable application delivery.
Answer:
A multi-stage Docker build uses multiple FROM instructions in a single Dockerfile.
It compiles and builds dependencies in a builder stage containing compilers and build tools. It then
copies only the compiled assets (like a virtual environment) to a clean runner stage that uses a minimal
base image (like python-slim ), discarding all compiler tools. This reduces the final image size and
minimizes security vulnerabilities.
Question 2: Why should you avoid running containers as the root user?
Answer:
By default, Docker container processes run as root. If an attacker exploits a code vulnerability (like
directory traversal or remote execution) inside the container, they can escape the container isolation
barrier and gain administrative control over the host system. Running containers as a non-root user
prevents container escape exploits.
13. Exercises
Write a CLI command to run a Docker container named web-api in detached mode, binding host port
80 to container port 8000 .
Solution:
This mini-project is a local orchestration template using Docker Compose. It sets up three services: a
FastAPI web service, a PostgreSQL database with persistent storage volumes, and a Redis cache
instance, mapping configurations dynamically using environment variables.
# [Link]
version: '3.8'
services:
# 1. FastAPI Application Service
web:
build:
context: .
dockerfile: Dockerfile
ports:
- "8000:8000"
environment:
- DATABASE_URL=postgresql://postgres:db_pass@db:5432/app_db
- REDIS_URL=redis://cache:6379/0
depends_on:
- db
- cache
volumes:
postgres_data:
Key Takeaways
1. Containerization isolates your application code, dependencies, and environment, preventing runtime
differences between development and production.
2. Use multi-stage builds to compile dependencies in a builder stage, copying only the virtual
environment to a clean runner stage to reduce image size and improve security.
3. Order Dockerfile commands (copying dependencies and running pip install before copying
source code) to leverage Docker's layer caching and speed up build times.
Further Reading
1. Introduction
When developing a FastAPI application, you run it using Uvicorn with reloading active: uvicorn
main:app --reload .
This setup is designed for development productivity. It polls files for changes and restarts the server
when code is saved.
However, this setup is not suitable for production. It runs on a single thread and single CPU core,
cannot recover from process crashes, and lacks protection against slow-client attacks.
In production, you must run your application using a Process Manager (like Gunicorn) that spawns
and monitors multiple worker processes (running Uvicorn worker classes), and place it behind a
Reverse Proxy (like Nginx).
In this chapter, we will learn how to configure Gunicorn and Uvicorn for production, optimize process
concurrency, and deploy secure Nginx reverse proxies.
2. Theory
1. Reverse Proxy (Nginx): The gateway. It handles SSL/TLS decryption, serves static files, rates
limits requests, and buffers slow client connections.
2. Process Manager (Gunicorn): Spawns, monitors, and restarts worker processes. If a worker
crashes due to a memory leak or segfault, Gunicorn spawns a new one instantly.
3. ASGI Workers (Uvicorn): The execution engines. Each worker runs a single-threaded event loop
that processes FastAPI request-response cycles.
To maximize CPU utilization on multi-core servers, run one worker process per CPU core.
If your production server has 4 CPU cores, you should configure Gunicorn to run 9 workers. This
ensures that even if some workers are waiting on disk I/O, other workers are active on the CPU.
3. Internal Working
In a Slowloris attack, an attacker opens hundreds of connections and sends HTTP requests extremely
slowly (e.g. 1 byte every 10 seconds).
If Uvicorn is exposed directly to the internet, all worker event loops will become blocked waiting for
these slow requests to complete, preventing legitimate users from accessing the API.
Placing Nginx in front of your ASGI server protects against this. Nginx is an event-driven, non-
blocking reverse proxy. It buffers the incoming request bytes until the complete payload is received,
and only forwards the request to your ASGI workers when it is ready.
4. API Reference
gunicorn main:app \
-w 4 \ # Number of workers
-k [Link] \ # Use Uvicorn ASGI worker class
-b [Link]:8000 # Bind to all interfaces on port 8000
5. Practical Examples
This configuration configures Nginx to act as a reverse proxy, terminated with SSL, routing traffic to a
Gunicorn socket.
# /etc/nginx/sites-available/[Link]
location / {
# 3. Forward traffic to Gunicorn listening on a local Unix socket
proxy_pass [Link]
# Buffer configurations
proxy_buffering on;
proxy_buffer_size 8k;
}
}
For local communication between Nginx and Gunicorn on the same server, bind Gunicorn to a Unix
Socket (e.g. unix:/tmp/[Link] ) instead of a TCP port ( [Link]:8000 ).
Unix sockets run within the OS kernel space, bypassing the TCP loopback network stack (avoiding
TCP handshakes and routing overhead), which improves throughput.
7. Common Mistakes
Mistake: Deploying Uvicorn directly on public ports ( 80 or 443 ) without a reverse proxy.
This exposes your application to slow-client Denial-of-Service (DoS) attacks, lacks SSL termination, and
slows down static file delivery.
Correction: Always place Nginx or an API Gateway in front of your ASGI server.
8. Performance Tips
Keep-Alive Tuning
Configure your reverse proxy and ASGI server to keep TCP connections alive ( keepalive timeouts).
Reusing established connections for subsequent requests saves TCP handshake overhead, reducing
latency.
9. Security Considerations
By default, Nginx and Gunicorn include version details in error headers (e.g. Server: nginx/1.18.0 ).
This helps attackers identify specific CVE vulnerabilities.
Check permission settings on the Unix socket file ( /tmp/[Link] ), ensuring Nginx has read
and write access.
Production systems deploy Gunicorn clusters across multiple nodes behind cloud load balancers (like
AWS ALB), providing automatic failover and scalability.
Answer:
Uvicorn is a fast ASGI server, but it lacks process management capabilities. If a worker process crashes
due to a memory leak or segfault, Uvicorn cannot restart it.
Gunicorn acts as a Process Manager. It monitors worker processes, restarts crashed workers, and
manages concurrency across multi-core processors. In production, Gunicorn coordinates process
management, while Uvicorn workers execute the ASGI code.
Question 2: Why place a reverse proxy (like Nginx) in front of an ASGI server?
Answer:
Nginx protects against slow-client attacks (like Slowloris) by buffering request bytes until the complete
payload is received before forwarding it to workers. It also handles SSL/TLS termination, serves static
assets efficiently, rates limits requests, and provides load balancing.
13. Exercises
Calculate the optimal number of Gunicorn workers for a production server with 8 CPU cores.
Solution:
workers = (2 × 8) + 1 = 17 workers
# gunicorn_conf.py
import multiprocessing
import os
# 1. Bind configurations
# If running inside a container stack, bind to TCP; otherwise use Unix socket
bind = [Link]("BIND", "unix:/tmp/[Link]")
worker_class = "[Link]"
# 3. Connection Tuning
backlog = 2048 # Max queued connections
timeout = 30 # Restart workers hanging for over 30s
keepalive = 5 # Keep-alive connection persistence timeout
# 4. Logging configurations
accesslog = "-" # Log access details to stdout
errorlog = "-" # Log error details to stderr
loglevel = [Link]("LOG_LEVEL", "info")
Key Takeaways
1. Gunicorn acts as a process manager, monitoring and spawning Uvicorn worker processes to
maximize CPU utilization.
Further Reading
1. Introduction
When multiple developers contribute code to the same repository, manual quality checks break down. A
developer might forget to run unit tests, ignore formatting guidelines, or introduce syntax errors that
break production deployments.
To solve this, we use CI/CD (Continuous Integration & Continuous Deployment) pipelines.
Continuous Integration (CI): Automatically triggers checks (linting, type checking, unit tests) every
time a developer pushes code or opens a pull request. If any check fails, the code cannot be
merged.
Continuous Deployment (CD): Automatically packages the validated code (e.g. compiling a
Docker image) and deploys it to staging or production servers once the code is merged.
In this chapter, we will learn how to design automated delivery pipelines, configure GitHub Actions for
FastAPI, and write linting, testing, and container build steps.
2. Theory
[Linting & Formatting] ---> [Type Checking (Mypy)] ---> [Test Execution (Pytest)] -
--> [Build & Deploy]
1. Linting & Formatting: Verifies code style guidelines and detects formatting anomalies. (Using fast
tools like ruff or black ).
2. Static Type Checking: Validates type hint contracts across modules. (Using mypy ).
3. Automated Testing: Executes the integration and unit test suite against mocked services. (Using
pytest ).
4. Build & Release: Compiles a production Docker image, pushes it to a container registry, and
notifies the cloud environment to restart services.
YAML Configurations
GitHub Actions workflows are defined in YAML files inside the .github/workflows/ directory.
Workflows are event-driven: they listen for events (like push or pull_request on the main branch) and
run a series of Jobs inside clean virtual environments (runners).
3. Internal Working
GitHub provisions a clean virtual machine runner (running Ubuntu, Windows, or macOS).
It sets up Python, restores dependency caches, executes test scripts, and reports statuses.
Once completed, the runner virtual machine is destroyed, ensuring zero environment state leakage
between jobs.
4. API Reference
on : The event that triggers the workflow (e.g., push: branches: [main] ).
jobs : A list of execution blocks that run in parallel or sequentially.
5. Practical Examples
This YAML file defines a complete CI workflow that automates linting, type validation, and test suites.
# .github/workflows/[Link]
name: Continuous Integration Pipeline
on:
push:
branches: [ main, develop ]
pull_request:
branches: [ main ]
jobs:
# Job 1: Quality Checks
quality-checks:
runs-on: ubuntu-latest
steps:
# 1. Checkout repository code
- name: Checkout Code
uses: actions/checkout@v3
# 3. Install packages
- name: Install Dependencies
run: |
python -m pip install --upgrade pip
pip install ruff mypy pytest httpx sqlalchemy aiosqlite
Installing dependencies on clean runners takes time, slowing down developer feedback.
uses: actions/setup-python@v4
with:
cache: 'pip'
This caches downloaded python packages. If your [Link] has not changed, the runner
restores packages from the cache in seconds, bypassing the download step.
7. Common Mistakes
Mistake: Committing production credentials, database keys, or AWS access tokens inside your
workflow YAML files.
Anyone with read access to the repository can steal these credentials.
Correction: Store credentials inside GitHub's Encrypted Secrets (Repository Settings -> Secrets and
Variables -> Actions). Reference them in your YAML file using environment variables:
env:
DATABASE_URL: ${{ secrets.PROD_DATABASE_URL }}
8. Performance Tips
Split linting checks and test execution into separate, independent jobs:
jobs:
linting:
runs-on: ubuntu-latest
steps: ...
testing:
runs-on: ubuntu-latest
steps: ...
GitHub Actions runs independent jobs in parallel on different runners, reducing the time developers wait
for feedback.
9. Security Considerations
Best Practice: Enforce read-only scopes by default inside your workflow configurations, granting
write permissions only to release/deployment jobs.
permissions:
contents: read
If a build fails:
Click on the failed run, select the job, and expand the failed step.
Pytest outputs and Python compilation errors are logged in the terminal stream, helping you locate
the bug.
Upon merging code to the main branch, a CD workflow compiles a production-grade Docker image,
tags it with the git commit hash, and pushes it to a container registry (like Docker Hub or AWS ECR),
triggering a deployment on Kubernetes.
Continuous Integration (CI) is the practice of automatically running linting, formatting, type
checking, and unit tests whenever developers push code changes, ensuring quality control before
merging.
Continuous Deployment (CD) is the practice of automatically packaging and deploying the
merged, validated code to staging or production environments, enabling fast and frequent releases.
Answer:
GitHub Actions runners start in clean virtual machines. Installing python dependencies from scratch on
every run takes time, slowing down feedback cycles. Caching dependencies saves downloaded
packages, allowing the runner to restore them in seconds if the dependencies file has not changed,
saving build time.
13. Exercises
Write the trigger section for a GitHub Actions workflow that executes ONLY on pull requests targeting
the main branch.
Solution:
on:
pull_request:
branches:
- main
This mini-project is a complete GitHub Actions production workflow script. It runs quality checks (linting,
types, pytest) and compiles a Docker image, logging in to a registry securely.
# .github/workflows/[Link]
name: Production Release Pipeline
on:
push:
tags:
- 'v*.*.*' # Triggers when a release tag is pushed, e.g., v1.0.0
jobs:
# Stage 1: Run Quality Checks
ci-verification:
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
# Stage 2: Build and Push Docker Image (Executes only if Stage 1 succeeds)
docker-release:
needs: ci-verification
runs-on: ubuntu-latest
steps:
- name: Checkout Code
uses: actions/checkout@v3
Key Takeaways
1. CI/CD pipelines automate code quality checks and deployments, reducing human error.
2. Configure GitHub Actions workflows inside .github/workflows/ using YAML configuration files.
4. Protect production credentials by storing them in GitHub's Encrypted Secrets. Never commit secrets
to version control.
Further Reading
1. Introduction
When an application crashes on your local machine, you can check the terminal output, add print
statements, or attach a debugger. In a production environment serving thousands of users across
multiple server instances, you cannot do this.
If a user reports that a checkout endpoint failed, you must be able to trace their request path and
identify the failure point.
This ability to understand the internal state of a system based on its external outputs is called
Observability.
2. Metrics: Numeric measurements over time (rate of requests, error counts, memory usage).
3. Traces: The path of a request as it flows through your system.
In this chapter, we will learn how to configure structured JSON logging using structlog , expose
metrics for Prometheus, and implement request tracing.
2. Theory
Structured Logging
In production, log files are processed by log aggregation tools (like Elasticsearch, Loki, or Datadog).
Problem: Parsing this text using regular expressions to extract metrics is slow and prone to
errors.
Solution: Log aggregation tools can parse and index JSON key-value pairs automatically,
allowing you to query, filter, and alert on specific fields.
Prometheus collects metrics from your API by polling (scraping) a /metrics endpoint. It supports three
core metric types:
3. Histogram: Measures the distribution of values (e.g. request latencies), allowing you to calculate
percentiles (like p95 or p99 response times).
3. Internal Working
To trace bugs in log streams, inject a unique Correlation ID (a random UUID) into the context of every
request.
A middleware generates a correlation ID for every incoming request and attaches it to the request
lifecycle.
Any log statement executed during that request includes this correlation ID.
If an error occurs, you can search your log aggregator for that correlation ID to view every log
statement associated with that request.
4. API Reference
# Initialize logger
logger = structlog.get_logger()
2. Prometheus Client
5. Practical Examples
This example configures a Prometheus scraper and exposes a /metrics endpoint in FastAPI.
app = FastAPI()
HTTP_REQUEST_DURATION = Histogram(
"http_request_duration_seconds",
"Histogram of request processing latencies."
)
start_time = time.perf_counter()
# Process request
response = await call_next(request)
# Calculate duration
duration = time.perf_counter() - start_time
# Record metrics
HTTP_REQUESTS_TOTAL.labels(
method=[Link],
endpoint=[Link],
status_code=response.status_code
).inc()
HTTP_REQUEST_DURATION.observe(duration)
return response
Avoid adding parameters with unbounded values (high cardinality) as labels in Prometheus metrics:
Bad: HTTP_REQUESTS_TOTAL.labels(user_id=[Link]).inc()
Good: HTTP_REQUESTS_TOTAL.labels(status_code=200).inc()
Prometheus creates a separate timeseries in memory for every unique combination of labels.
Adding high-cardinality values (like user IDs or emails) will exhaust Prometheus memory, crashing
the metrics engine.
7. Common Mistakes
8. Performance Tips
Writing logs to disk is a blocking I/O operation. In production, configure your logging framework to write
to standard output ( stdout ) asynchronously, allowing the host container engine (like Docker) to redirect
and buffer the log stream.
9. Security Considerations
If you log unvalidated user input directly as text, an attacker can insert carriage return and line feed
(CRLF) characters, appending fake log entries to fool security auditors.
Mitigation: Using structured JSON logging avoids this vulnerability, as CRLF characters are
encoded as standard JSON string fields rather than system line breaks.
Structlog supports conditional rendering: configure your logger to output formatted, colorized text in
development, and transition to structured JSON outputs in production environments.
Answer:
Structured logging outputs log entries in a structured machine-readable format (usually JSON) rather
than plain text.
Log aggregation systems (like Datadog or Elasticsearch) parse and index JSON key-value pairs
automatically. This allows engineers to query and filter logs by specific fields (like searching for a
specific user_id ), which is slow and error-prone when parsing unstructured text using regular
expressions.
Question 2: Why should you avoid using user emails or IDs as labels in Prometheus
metrics?
Answer:
Prometheus creates a separate timeseries database record in memory for every unique combination of
metric label values.
Using high-cardinality values (like user IDs or emails) creates millions of unique timeseries records. This
exhausts Prometheus memory, crashing the metrics collection server. Labels should be restricted to
low-cardinality values (like HTTP methods or status codes).
13. Exercises
Solution:
USER_REGISTRATIONS_TOTAL = Counter(
"user_registrations_total",
"Total count of user registrations."
)
This mini-project implements a complete observability module for FastAPI. It configures structured
JSON logging with request correlation IDs, tracks request latencies and status code rates using
Prometheus, and exposes a secure metrics scraping route.
import uuid
import time
import structlog
from fastapi import FastAPI, Request, Response
from prometheus_client import Counter, Histogram, make_asgi_app
REQUEST_DURATION_HISTOGRAM = Histogram(
"api_request_duration_seconds",
"Request latencies distribution."
)
# 3. Observability Middleware
@[Link]("http")
async def observability_middleware(request: Request, call_next):
if [Link] == "/metrics":
return await call_next(request)
start_time = time.perf_counter()
try:
response: Response = await call_next(request)
duration = time.perf_counter() - start_time
# Track metrics
HTTP_REQUESTS_COUNTER.labels(
method=[Link],
path=[Link],
status=response.status_code
).inc()
REQUEST_DURATION_HISTOGRAM.observe(duration)
# Log completion
[Link](
"request_completed",
status_code=response.status_code,
duration_ms=round(duration * 1000, 2)
)
return response
except Exception as exc:
duration = time.perf_counter() - start_time
HTTP_REQUESTS_COUNTER.labels(
method=[Link],
path=[Link],
status=500
).inc()
Key Takeaways
1. Observability uses logs, metrics, and traces to monitor and troubleshoot application health.
2. Structured JSON logging enables log aggregation tools to index and query log statements efficiently.
3. Expose API request metrics (counters, histograms) to Prometheus, avoiding high-cardinality labels
(like user IDs) to prevent memory exhaustion.
4. Use Correlation IDs (Trace IDs) to link log statements across request lifecycles.
Further Reading
1. Introduction
In simple FastAPI tutorials, endpoints query databases directly using SQLAlchemy and return the
results:
@[Link]("/users/{id}")
def get_user(id: int, db: Session = Depends(get_db)):
return [Link](User).filter([Link] == id).first()
For small apps, this is fine. However, as applications grow, this coupling becomes problematic:
Business Logic Leakage: Business rules (like calculating discounts or validating account
balances) bleed into route controllers, leading to duplicated code.
Database Dependency: Your core business rules depend directly on database schemas. If you
change database drivers or transition to a NoSQL database, you have to rewrite your entire
codebase.
Testing Obstacles: Testing a business rule requires running real database queries, slowing down
tests.
To build large, maintainable applications, we apply Architectural Patterns like Domain-Driven Design
(DDD) and Clean (Hexagonal) Architecture.
In this chapter, we will learn how to isolate business logic from database layers, implement the
Repository Pattern, and decouple FastAPI route handlers from database frameworks.
2. Theory
The core rule of Clean Architecture is the Dependency Rule: Inner layers must not know anything
about outer layers.
+-------------------------------------------------+
| Delivery Layer (Adapters: FastAPI HTTP, CLI) |
| +---------------------------------+ |
| | Application Layer (Use Cases) | |
| | +-----------------+ | |
| | | Domain Layer | | |
| | | (Pure Entities)| | |
| | +-----------------+ | |
| +---------------------------------+ |
+-------------------------------------------------+
| Infrastructure Layer (Adapters: DB, SMTP, S3) |
+-------------------------------------------------+
1. Domain Layer (Center): Contains pure business rules, entities, and value objects. It has zero
dependencies on external libraries (like FastAPI, SQLAlchemy, or Pydantic).
2. Application Layer: Orchestrates use cases (e.g. RegisterUserUseCase ). It depends on the Domain
layer, but is unaware of specific databases or frameworks.
3. Adapters (Infrastructure & Delivery): The outer ring. FastAPI endpoints (Delivery) and database
queries (Infrastructure) act as adapters that interact with the inner application layer.
Entities: Domain objects defined by a unique identity that persists over time (e.g. a User with a
unique ID).
Value Objects: Immutable objects defined solely by their attributes (e.g. an Address or an
EmailAddress ). They have no identity. If two addresses have the same fields, they are identical.
Repositories: An abstraction interface that behaves like an in-memory collection of domain objects.
It defines CRUD signatures without committing to a specific database implementation.
3. Internal Working
Clean architecture relies on Dependency Inversion: high-level business logic should not depend on
low-level database operations; both should depend on abstractions.
In Python, we implement this using Abstract Base Classes (ABCs):
4. API Reference
Subclasses must override all methods decorated with @abstractmethod before they can be
instantiated, enforcing code contracts at runtime.
5. Practical Examples
This example illustrates a pure domain entity containing business validation logic, independent of any
database annotations or API routers.
class BankAccount:
"""Pure domain entity containing business rules."""
def __init__(self, account_id: str, owner: str, balance: float = 0.0):
[Link] = account_id
[Link] = owner
self._balance = balance
@property
def balance(self) -> float:
return self._balance
Do not use SQLAlchemy classes (which contain database-specific decorations like __tablename__ or
mapped_column ) as your core domain models.
Best Practice: Keep your domain classes as pure Python classes. Use a Repository class to map
data between your pure domain models and your database-specific SQLAlchemy models. This
isolates your business rules from database schema updates.
7. Common Mistakes
Mistake: Accessing database sessions ( [Link]() ) or running SQL queries inside your
business logic classes.
If you couple your domain models to databases, you cannot run unit tests without database
connections, and refactoring schemas becomes difficult.
Correction: Execute database commits in the application layer or database repositories, keeping the
domain layer database-agnostic.
8. Performance Tips
By using abstract repository contracts, you can write an in-memory repository implementation (using a
Python dictionary) for testing:
class MockUserRepository(UserRepository):
def __init__(self):
[Link] = {}
def save(self, user):
[Link][[Link]] = user
This allows your unit tests to run in microseconds without database overhead.
9. Security Considerations
1. Invariant Enforcement
Enforce your business security rules inside the constructor or methods of your domain model entities
(e.g. verifying that a bank account balance cannot be negative). This guarantees that the entity can
never enter an invalid state, regardless of database or API issues.
If your application returns database mapping errors, check your repository layer. Ensure that database
models are mapped to domain models correctly when retrieving data, and that domain models are
mapped to database schemas when saving.
If a company decides to migrate from PostgreSQL to MongoDB, a clean architecture allows developers
to rewrite only the database repository implementations. The core business rules and API endpoints
remain unchanged, saving development time.
Answer:
The Dependency Rule states that source code dependencies must only point inwards.
Inner layers (like the Domain layer containing business rules) must be completely independent of outer
layers (like databases, frameworks, or APIs). The Domain layer knows nothing about FastAPI or
SQLAlchemy, allowing business logic to remain stable when databases or frameworks are updated.
Answer:
The Repository Pattern decouples the application layer from the database layer by exposing an
interface that acts like an in-memory collection of domain objects.
It defines CRUD method signatures without committing to a specific database implementation. This
allows developers to mock databases during testing or switch databases (e.g. from PostgreSQL to
MongoDB) without modifying business logic.
13. Exercises
Define an abstract repository interface named ItemRepository containing two methods: get_by_id(id:
str) and save(item: Item) .
Solution:
class ItemRepository(ABC):
@abstractmethod
def get_by_id(self, item_id: str):
pass
@abstractmethod
def save(self, item) -> None:
pass
This mini-project is a banking API structured using Clean Architecture. It defines a pure Domain Model,
an abstract Repository contract, a mock in-memory implementation, an Application Service use case,
and a FastAPI endpoint.
# =========================================================
# 1. DOMAIN LAYER (Pure Python - No External Dependencies)
# =========================================================
class Wallet:
def __init__(self, wallet_id: str, owner: str, balance: float = 0.0):
[Link] = wallet_id
[Link] = owner
[Link] = balance
# =========================================================
# 2. PORT / INTERFACE LAYER (Contracts)
# =========================================================
class WalletRepository(ABC):
@abstractmethod
def get(self, wallet_id: str) -> Wallet | None:
pass
@abstractmethod
def save(self, wallet: Wallet) -> None:
pass
# =========================================================
# 3. INFRASTRUCTURE LAYER (Adapters: In-Memory DB)
# =========================================================
class InMemoryWalletRepository(WalletRepository):
def __init__(self):
self._store = {}
# =========================================================
# 4. APPLICATION LAYER (Business Use Case orchestration)
# =========================================================
class DepositUseCase:
def __init__(self, repo: WalletRepository):
[Link] = repo
# =========================================================
# 5. DELIVERY LAYER (FastAPI Routes)
# =========================================================
app = FastAPI(title="Decoupled Banking Portal")
class DepositRequest(BaseModel):
wallet_id: str
amount: float
@[Link]("/wallets/deposit")
async def deposit_funds(
payload: DepositRequest,
repo: Annotated[WalletRepository, Depends(get_wallet_repository)]
):
use_case = DepositUseCase(repo)
try:
updated_wallet = use_case.execute(payload.wallet_id, [Link])
return {
"wallet_id": updated_wallet.id,
"owner": updated_wallet.owner,
"balance": updated_wallet.balance
}
except ValueError as ve:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=str(ve)
)
Key Takeaways
1. Clean Architecture decouples business rules from databases, frameworks, and APIs.
2. The Domain Layer sits at the center, containing pure Python classes (entities, value objects) with
zero external dependencies.
3. The Repository Pattern uses Abstract Base Classes (ABCs) to define database contracts, allowing
developers to switch database engines or mock data during testing easily.
4. Keep FastAPI endpoints thin. Endpoints should act as delivery adapters that receive request inputs,
execute application use cases, and return outputs.
Further Reading
Clean Architecture: A Craftsman's Guide to Software Structure and Design by Robert C. Martin
(Prentice Hall)
Domain-Driven Design: Tackling Complexity in the Heart of Software by Eric Evans (Addison-
Wesley)
Architecture Patterns in Python ([Link]
1. Introduction
FastAPI is inherently fast, but as request volumes scale to thousands of requests per second, small
inefficiencies in your application code will compound. A blocking socket operation, a slow JSON
serialization loop, or memory leaks inside custom middlewares will degrade performance and increase
infrastructure costs.
Optimizing performance should never be based on guesswork. You must follow a systematic approach:
In this chapter, we will learn how to profile FastAPI applications using cProfile , optimize JSON
parsing, leverage concurrency via asyncio Task Groups, and scale database connection pools.
2. Theory
cProfile is Python's built-in execution profiler. It measures the execution time and call counts of every
function in your code. By running your application under a profiler while executing load tests, you can
compile a detailed report showing which functions occupy the most CPU time.
A common mistake when calling multiple external services (e.g. fetching user profiles, billing status, and
inventory data) is executing requests sequentially:
By utilizing [Link] or modern Python 3.11+ Task Groups, you can run these queries
concurrently on the event loop, reducing the total latency to the duration of the slowest query:
3. Internal Working
In data-heavy APIs, JSON serialization (converting Python dictionaries or Pydantic models to JSON
strings) is a major CPU bottleneck.
FastAPI uses Python's standard json library by default, which is written in Python and is slow.
Orjson: A fast, C-based JSON library written in Rust. It serializes datatypes (like datetimes and
decimals) up to 10x faster than the standard library, significantly reducing CPU overhead.
4. API Reference
To use orjson as the default serializer, inject the custom ORJSONResponse class into your routers or
application:
5. Practical Examples
This example demonstrates how to perform multiple async fetches concurrently using Python 3.11+
TaskGroup context managers.
import asyncio
from fastapi import FastAPI
import httpx
app = FastAPI()
@[Link]("/aggregate")
async def aggregate_data():
async with [Link]() as client:
# Create Task Group
async with [Link]() as tg:
# 1. Spawn concurrent tasks
task_a = tg.create_task(fetch_service_a(client))
task_b = tg.create_task(fetch_service_b(client))
In production, database connection setup (TCP handshakes, authentication) takes time. Never open
and close database connections on every request.
Best Practice: Set appropriate connection pool limits ( pool_size and max_overflow in
SQLAlchemy) to keep a warm pool of active connections, reducing connection overhead.
7. Common Mistakes
Correction: Return database objects directly and leverage FastAPI's response_model type hints, or
use orjson response classes to handle serialization efficiently.
8. Performance Tips
9. Security Considerations
Middlewares run on every request. If your middleware appends data to global lists or holds reference
pointers to request scopes without releasing them, Python's Garbage Collector cannot clean them up,
leading to a memory leak that will eventually crash your container.
Mitigation: Keep middlewares stateless. Never store request-specific objects in global scopes or
class variables.
Open the profile file using visualization tools (like Snakeviz) to view call trees and locate execution
hotspots.
Financial tickers query stock rates, user portfolios, and news feeds concurrently using Task Groups.
This aggregates data quickly and keeps latency low, ensuring real-time delivery.
Answer:
Sequential execution runs operations one after another, summing the latencies of all calls.
Task Groups execute operations concurrently on the event loop. The total latency is limited only by the
slowest query, significantly reducing response times.
13. Exercises
Write an async function fetch_all(urls: list[str]) that queries a list of URLs concurrently using
[Link] .
Solution:
import asyncio
import httpx
This mini-project is a performance testing module. It executes and measures sequential vs. concurrent
task group runs, compiling execution times and verifying optimization benefits.
import asyncio
import time
from fastapi import FastAPI
import httpx
app = FastAPI()
class BenchmarkRunner:
@staticmethod
async def run_sequential() -> float:
start = time.perf_counter()
# Sequential runs: Latencies sum up
await simulate_slow_network_call(0.2)
await simulate_slow_network_call(0.2)
await simulate_slow_network_call(0.2)
return time.perf_counter() - start
@staticmethod
async def run_concurrent() -> float:
start = time.perf_counter()
# Concurrent runs using Task Group: Latencies run in parallel
async with [Link]() as tg:
tg.create_task(simulate_slow_network_call(0.2))
tg.create_task(simulate_slow_network_call(0.2))
tg.create_task(simulate_slow_network_call(0.2))
return time.perf_counter() - start
@[Link]("/benchmark")
async def run_performance_test():
seq_time = await BenchmarkRunner.run_sequential()
con_time = await BenchmarkRunner.run_concurrent()
return {
"sequential_duration_seconds": round(seq_time, 4),
"concurrent_duration_seconds": round(con_time, 4),
"latency_reduction_percentage": round(improvement, 2)
}
Key Takeaways
1. Optimize performance systematically: profile first, locate bottlenecks, and apply optimizations.
2. Use Python 3.11+ Task Groups or [Link] to execute independent tasks concurrently,
reducing latency.
3. Configure ORJSONResponse in FastAPI to use the fast, Rust-based JSON serializer, reducing CPU
load.
4. Keep connection pools warm and middlewares stateless to prevent connection overhead and
memory leaks in production.
Further Reading
1. Introduction
Building a working API prototype is easy. Running that same API at scale under heavy user load with
zero downtime is a different challenge.
Before a FastAPI service is deployed to production, it must pass strict quality controls. If it lacks health
checks, container engines (like Kubernetes) cannot monitor its status. If credentials are leaked, the
system is exposed to breach. If connection limits are unconfigured, database queries will freeze during
traffic spikes.
Production Readiness is the process of auditing and hardening an application across security,
database stability, configuration, and monitoring metrics.
In this chapter, we will walk through the ultimate production deployment checklist, learn the difference
between liveness and readiness probes, and build a resilient health check system.
2. Theory
Container orchestrators (like Kubernetes) use HTTP endpoints to monitor container health:
Behavior: If this endpoint fails (returns HTTP 500), the orchestrator assumes the container has
crashed or hung, and terminates and restarts the container.
Checks: Keeps checks simple. Do not query database connections here (if the database is
down, restarting the web container will not fix it).
Behavior: If this endpoint fails, the orchestrator stops routing user requests to this container
instance, redirecting traffic to other replicas.
Checks: Queries critical dependencies (PostgreSQL database connections, Redis cache
instances, active message queues).
3. Internal Working
If your health check queries the database without timeout parameters, the health check request will
hang indefinitely. This blocks the checker probe, causing the orchestrator to assume the container is
dead and restart it, which can trigger a restart loop across your cluster.
Always enforce strict query and connection timeouts (e.g. 2-5 seconds) inside your health check
logic.
4. API Reference
5. Practical Examples
This example implements liveness and readiness endpoints, querying database and cache health in
parallel.
import asyncio
from fastapi import FastAPI, HTTPException, status
from [Link] import text
from [Link] import AsyncSession
import [Link] as aioredis
from typing import Annotated
from fastapi import Depends
app = FastAPI()
# Simulated database dependency
async def get_db_session() -> AsyncSession:
raise NotImplementedError()
2. Databases
7. Common Mistakes
Mistake: Querying external payment APIs or SMS gateways inside your readiness check.
If the third-party service experiences a brief outage, your readiness check will fail, causing the
orchestrator to pull your healthy containers out of the load balancer, taking down your entire application.
Correction: Only query critical resources under your control (primary databases, caches, message
queues).
8. Performance Tips
If your readiness check is called frequently (e.g. every 5 seconds by multiple container replicas),
running a test query on your database can increase database load.
Optimization: Cache readiness test results in memory for a short period (e.g. 2-3 seconds) to
reduce query overhead on your database.
9. Security Considerations
Expose endpoints like /metrics or /healthz/readiness strictly to internal network ranges (e.g., your
Kubernetes cluster IP block). Exposing metrics publicly allows attackers to map your database names,
query throughput, and traffic spikes, which can be used to plan targeted attacks.
10. Debugging Techniques
Verify that the readiness checks fail (returning HTTP 503) and the load balancer stops routing traffic
to the container.
Verify that the liveness checks continue to pass (returning HTTP 200), preventing unnecessary
container restarts.
Enterprise companies run readiness audits to prove to auditors that their APIs are secure, monitored,
and feature automated failover, ensuring compliance with security standards.
Question 1: What is the difference between a Liveness Probe and a Readiness Probe?
Answer:
Liveness Probes verify if the container process is running. If it fails, the orchestrator assumes the
container is dead and restarts it. It should be kept simple.
Readiness Probes check if the application is ready to accept user requests. If it fails, the
orchestrator stops sending traffic to the container but does not restart it. It should query critical
resources like databases and caches.
Question 2: Why should you avoid querying external third-party APIs during readiness
checks?
Answer:
Readiness checks should only monitor resources within your control. If you query an external third-party
API and it experiences an outage, your readiness check will fail, causing the orchestrator to pull your
healthy web containers out of rotation, taking down your entire application.
13. Exercises
Write a fast, non-blocking liveness route in FastAPI that returns status code 200 with zero external
calls.
Solution:
app = FastAPI()
@[Link]("/healthz")
async def healthz():
return {"status": "ok"}
This mini-project implements a production-grade readiness audit service. It checks database and cache
connections in parallel, logs detailed structural reports, and handles timeouts and dependency failures
gracefully.
import asyncio
import time
import structlog
from fastapi import FastAPI, HTTPException, status
from pydantic import BaseModel
class HealthStatus(BaseModel):
status: str
db_connected: bool
cache_connected: bool
duration_ms: float
class EnterpriseHealthAuditor:
@classmethod
async def perform_audit(cls) -> HealthStatus:
start_time = time.perf_counter()
# Execute checks concurrently with timeouts
try:
async with [Link](2.0):
# Run tasks in parallel
db_task = asyncio.create_task(check_db_health())
cache_task = asyncio.create_task(check_cache_health())
return HealthStatus(
status=overall_status,
db_connected=db_ok,
cache_connected=cache_ok,
duration_ms=round(duration, 2)
)
@[Link]("/healthz/readiness", response_model=HealthStatus)
async def readiness_probe():
status_report = await EnterpriseHealthAuditor.perform_audit()
if status_report.status != "healthy":
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=status_report.dict()
)
return status_report
Key Takeaways
1. Liveness probes verify if the container is running and should be kept simple. Readiness probes
check if the application is ready to accept traffic by verifying critical resources (databases, caches).
2. Set timeouts (e.g. 2 seconds) on all dependency checks to prevent probes from hanging and
causing restart loops.
3. Restrict access to metrics and health checks to internal network ranges to protect system
information.
4. Verify your security configurations, database connections, and logging settings against the
production checklist before deploying updates.
Further Reading
1. Introduction
Throughout this handbook, we have explored FastAPI's features, database integration layers,
concurrency configurations, security standards, and deployment architectures.
To consolidate these concepts, this final chapter walks through the system design and implementation
of two real-world case studies:
1. Real-time Chat & Notification Engine: A scalable WebSocket service using Redis Pub/Sub for
horizontal broadcasting.
2. High-throughput File Processing & Compression Pipeline: A secure upload workflow integrating
S3 storage and asynchronous Celery worker processes.
These case studies represent standard enterprise architectures, demonstrating how to design and build
scalable, production-ready systems.
System Requirements
Architecture Design
Client A
FastAPI Server 1
Unsupported markdown:
list
FastAPI Server 2
Client B
This script runs a scalable WebSocket server that handles authentication, manages connection
lifecycles, and broadcasts messages across multiple nodes using Redis.
import asyncio
import jwt
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Query, HTTPException,
status
import [Link] as aioredis
from typing import List
# Configuration Keys
SECRET_KEY = "chat-system-secret-key"
ALGORITHM = "HS256"
REDIS_URL = "redis://localhost:6379/0"
class WebSocketManager:
def __init__(self):
self.local_connections: List[WebSocket] = []
manager = WebSocketManager()
@app.on_event("startup")
async def startup_event():
# Start the background Redis listener task
redis_client = [Link](connection_pool=redis_pool)
[Link].redis_pub = redis_client
asyncio.create_task(redis_listener_task_wrapper(redis_client))
await [Link](websocket)
redis_pub: [Link] = [Link].redis_pub
try:
while True:
# Receive client message
data = await websocket.receive_text()
# Publish message to Redis, routing it to all active servers
await redis_pub.publish("global_chat", f"{username}: {data}")
except WebSocketDisconnect:
[Link](websocket)
# Broadcast leave message globally
await redis_pub.publish("global_chat", f"[System] {username} left the chat
room.")
System Requirements
Allow clients to upload large media files without blocking the API.
Architecture Design
Client
FastAPI Server
Unsupported markdown:
list
RabbitMQ Broker
This architecture defines a Celery task class for image/video compression, connects to an S3 bucket,
and provides FastAPI endpoints to schedule tasks and poll execution status.
# In [Link]
import time
import boto3
from celery import Celery
# Configure Celery
celery_app = Celery(
"file_pipeline",
broker="pyamqp://guest:guest@localhost:5672//",
backend="redis://localhost:6379/0"
)
# S3 Configuration
S3_BUCKET = "my-corporate-media-bucket"
s3_client = [Link]("s3")
@celery_app.task(bind=True, max_retries=3)
def process_media_compression(self, file_key: str) -> dict:
"""Downloads an asset from S3, processes it, and uploads the compressed
version."""
print(f"[Worker] Fetching object {file_key} from S3...")
# 1. Download file
local_input = f"/tmp/{file_key.split('/')[-1]}"
s3_client.download_file(S3_BUCKET, file_key, local_input)
output_key = f"optimized/{file_key.split('/')[-1]}"
s3_client.upload_file(local_output, S3_BUCKET, output_key)
return {
"status": "success",
"original_key": file_key,
"compressed_key": output_key
}
# In [Link]
import uuid
import asyncio
from fastapi import FastAPI, UploadFile, File, Depends, HTTPException
from worker import process_media_compression, celery_app, s3_client, S3_BUCKET
from [Link] import AsyncResult
@[Link]("/media/process", status_code=202)
async def process_media(file: UploadFile = File(...)):
# 1. Generate a unique key and upload file to S3
file_key = f"raw/{uuid.uuid4()}-{[Link]}"
await upload_file_to_s3(file, file_key)
return {
"task_id": [Link],
"status": "queued",
"raw_location": file_key
}
@[Link]("/media/status/{task_id}")
async def get_processing_status(task_id: str):
# 3. Query the Redis backend for task status
task_result = AsyncResult(task_id, app=celery_app)
response = {
"task_id": task_id,
"status": task_result.status
}
if task_result.status == "PROGRESS":
response["progress"] = task_result.[Link]("progress")
elif task_result.status == "SUCCESS":
response["result"] = task_result.result
elif task_result.status == "FAILURE":
response["error"] = str(task_result.info)
return response
Throughout this book, we have explored the complete backend development lifecycle from first
principles:
The Foundations: Python type hints, cooperative multitasking, the event loop, and ASGI
specifications.
The Framework: Routing compilation, request lifecycles, middlewares, dependency injection, and
Pydantic validation.
Security & Hardening: JWT verification, refresh token rotation, RBAC scopes, and defense against
OWASP vulnerabilities.
Enterprise Operations: Unit/integration testing, multi-stage Docker compilation, Nginx proxies, and
Prometheus structured metrics.
By applying these patterns, you can build scalable, secure, and resilient web applications that perform
under production-level loads. Happy coding!