FastAPI has become the go-to Python framework for building REST APIs in 2025 and 2026. It surpassed Flask in GitHub stars during 2025 — a milestone that reflects both the framework's technical quality and the community's shift toward type-safe, async-first Python. This tutorial walks you through building a complete, production-quality REST API: a Users and Items service with async PostgreSQL, JWT authentication, Pydantic v2 validation, Alembic migrations, and Docker Compose deployment. Every section includes working code you can run immediately.
Sources used throughout: FastAPI official documentation, SQLAlchemy 2.0 docs, Pydantic v2 docs, and TestDriven.io FastAPI guide.
Why FastAPI Surpassed Flask in 2025
Flask has been the backbone of Python web APIs for over a decade. So why did FastAPI overtake it in GitHub stars? Three reasons stand out:
1. Type safety built in. FastAPI uses Python type hints as its primary interface. Parameters, request bodies, and responses are all declared as typed Python objects. Editors provide full autocomplete; runtime catches mismatches before they reach production.
2. Automatic OpenAPI docs. FastAPI generates interactive Swagger UI (/docs) and ReDoc (/redoc) endpoints from your code with zero extra work. Your API is self-documenting by default — a massive productivity gain for teams.
3. Async-first performance. Built on Starlette and Uvicorn, FastAPI supports async def route handlers natively. Paired with an async database driver, a single process can handle thousands of concurrent requests that would block a traditional Flask or Django view.
Flask is a fine framework, but it requires third-party extensions for every one of these features — and those extensions do not always stay in sync. FastAPI ships them all coherently from day one.
Project Structure
Before writing a single line of code, establish a clean directory layout. A well-structured project is the difference between a prototype and a service you can hand off to another engineer.
fastapi-app/
├── app/
│ ├── __init__.py
│ ├── main.py # FastAPI application factory
│ ├── config.py # Settings via pydantic-settings
│ ├── database.py # Async SQLAlchemy engine + session
│ ├── models/
│ │ ├── __init__.py
│ │ ├── user.py # SQLAlchemy ORM model
│ │ └── item.py
│ ├── schemas/
│ │ ├── __init__.py
│ │ ├── user.py # Pydantic v2 schemas
│ │ └── item.py
│ ├── routers/
│ │ ├── __init__.py
│ │ ├── auth.py # Login / token endpoint
│ │ ├── users.py # CRUD routes for users
│ │ └── items.py # CRUD routes for items
│ ├── dependencies.py # Shared Depends() callables
│ └── security.py # JWT helpers
├── alembic/
│ ├── env.py
│ └── versions/
├── tests/
│ ├── conftest.py
│ ├── test_auth.py
│ └── test_items.py
├── alembic.ini
├── docker-compose.yml
├── Dockerfile
└── requirements.txt
This layout separates concerns cleanly: models live in one package, Pydantic schemas in another, route handlers in a third. Nothing is circular, and every layer is independently testable.
Dependencies
# requirements.txt
fastapi==0.115.0
uvicorn[standard]==0.30.1
sqlalchemy==2.0.30
asyncpg==0.29.0
alembic==1.13.1
pydantic==2.7.1
pydantic-settings==2.2.1
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
python-multipart==0.0.9
httpx==0.27.0
pytest==8.2.0
pytest-asyncio==0.23.6
Install with:
python -m venv venv && source venv/bin/activate
pip install -r requirements.txt
Configuration with Pydantic Settings
Pydantic v2's pydantic-settings package reads environment variables (or a .env file) and validates them at startup. This gives you typed configuration with zero boilerplate.
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8")
database_url: str = "postgresql+asyncpg://user:password@localhost:5432/appdb"
secret_key: str = "change-me-in-production"
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
settings = Settings()
Create a .env file (never commit it):
DATABASE_URL=postgresql+asyncpg://appuser:apppass@db:5432/appdb
SECRET_KEY=a-very-long-random-secret-key-here
Async Database with SQLAlchemy 2.0
SQLAlchemy 2.0 introduced a fully async session via AsyncSession and create_async_engine. This is the recommended pattern for FastAPI as of 2024–2026.
# app/database.py
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from app.config import settings
engine = create_async_engine(settings.database_url, echo=False)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
async def get_db() -> AsyncSession:
async with AsyncSessionLocal() as session:
yield session
get_db is an async generator that FastAPI's dependency injection system will call automatically for every request that needs a database session. The session is closed (and the connection returned to the pool) as soon as the request completes — even if an exception is raised.
ORM Models
# app/models/user.py
from datetime import datetime
from sqlalchemy import Boolean, DateTime, String, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
email: Mapped[str] = mapped_column(String(255), unique=True, index=True, nullable=False)
hashed_password: Mapped[str] = mapped_column(String(255), nullable=False)
is_active: Mapped[bool] = mapped_column(Boolean, default=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
items: Mapped[list["Item"]] = relationship("Item", back_populates="owner", lazy="selectin")
# app/models/item.py
from sqlalchemy import ForeignKey, String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.database import Base
class Item(Base):
__tablename__ = "items"
id: Mapped[int] = mapped_column(primary_key=True, index=True)
title: Mapped[str] = mapped_column(String(255), nullable=False, index=True)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
owner_id: Mapped[int] = mapped_column(ForeignKey("users.id"), nullable=False)
owner: Mapped["User"] = relationship("User", back_populates="items")
SQLAlchemy 2.0's Mapped annotation style makes column types explicit and editor-friendly. The mapped_column approach replaces the older Column(...) syntax and integrates seamlessly with Python's type system.
Pydantic v2 Schemas
Pydantic v2 (released mid-2023) is significantly faster than v1 and introduces a cleaner API. The key change: orm_mode = True is now model_config = ConfigDict(from_attributes=True).
# app/schemas/user.py
from pydantic import BaseModel, ConfigDict, EmailStr
class UserBase(BaseModel):
email: EmailStr
class UserCreate(UserBase):
password: str
class UserRead(UserBase):
model_config = ConfigDict(from_attributes=True)
id: int
is_active: bool
class UserReadWithItems(UserRead):
items: list["ItemRead"] = []
# app/schemas/item.py
from pydantic import BaseModel, ConfigDict
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class ItemUpdate(ItemBase):
title: str | None = None
class ItemRead(ItemBase):
model_config = ConfigDict(from_attributes=True)
id: int
owner_id: int
The from_attributes=True config flag allows Pydantic to read data from ORM model instances (attribute access) rather than requiring a dict. This is what makes UserRead.model_validate(user_orm_object) work without manual conversion.
JWT Authentication
Security Helpers
# app/security.py
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.config import settings
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def hash_password(password: str) -> str:
return pwd_context.hash(password)
def verify_password(plain_password: str, hashed_password: str) -> bool:
return pwd_context.verify(plain_password, hashed_password)
def create_access_token(subject: str | int, expires_delta: timedelta | None = None) -> str:
expire = datetime.now(timezone.utc) + (expires_delta or timedelta(minutes=settings.access_token_expire_minutes))
to_encode = {"sub": str(subject), "exp": expire}
return jwt.encode(to_encode, settings.secret_key, algorithm=settings.algorithm)
def decode_access_token(token: str) -> str:
"""Return the subject (user id) from a valid token, or raise JWTError."""
payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
sub: str | None = payload.get("sub")
if sub is None:
raise JWTError("Token has no subject")
return sub
python-jose handles JWT encoding/decoding. passlib with the bcrypt backend hashes passwords securely. Never store plain-text passwords; never store the JWT secret in source control.
Dependency Injection with FastAPI's Depends
FastAPI's dependency injection system is one of its most powerful features. A dependency is simply a callable that FastAPI calls before your route handler, injecting the return value as a parameter. Dependencies can themselves declare dependencies, forming a clean tree.
# app/dependencies.py
from typing import Annotated
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.models.user import User
from app.security import decode_access_token
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/token")
DBSession = Annotated[AsyncSession, Depends(get_db)]
TokenStr = Annotated[str, Depends(oauth2_scheme)]
async def get_current_user(token: TokenStr, db: DBSession) -> User:
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
user_id = decode_access_token(token)
except JWTError:
raise credentials_exception
result = await db.execute(select(User).where(User.id == int(user_id)))
user = result.scalar_one_or_none()
if user is None or not user.is_active:
raise credentials_exception
return user
CurrentUser = Annotated[User, Depends(get_current_user)]
Using Annotated aliases (DBSession, CurrentUser) keeps route signatures clean and removes repetitive Depends(...) calls from every endpoint. This is the pattern recommended in the FastAPI docs as of version 0.100+.
Auth Router
# app/routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserRead
from app.security import create_access_token, hash_password, verify_password
router = APIRouter(prefix="/auth", tags=["auth"])
@router.post("/register", response_model=UserRead, status_code=status.HTTP_201_CREATED)
async def register(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.email == user_in.email))
if result.scalar_one_or_none():
raise HTTPException(status_code=400, detail="Email already registered")
user = User(email=user_in.email, hashed_password=hash_password(user_in.password))
db.add(user)
await db.commit()
await db.refresh(user)
return user
@router.post("/token")
async def login(form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User).where(User.email == form_data.username))
user = result.scalar_one_or_none()
if not user or not verify_password(form_data.password, user.hashed_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
token = create_access_token(subject=user.id)
return {"access_token": token, "token_type": "bearer"}
The /auth/token endpoint accepts application/x-www-form-urlencoded (standard OAuth2 password flow), making it directly compatible with the Swagger UI "Authorize" button.
Users Router (Full CRUD)
# app/routers/users.py
from fastapi import APIRouter, HTTPException, status
from sqlalchemy import select
from app.dependencies import CurrentUser, DBSession
from app.models.user import User
from app.schemas.user import UserRead, UserReadWithItems
router = APIRouter(prefix="/users", tags=["users"])
@router.get("/me", response_model=UserReadWithItems)
async def read_current_user(current_user: CurrentUser):
return current_user
@router.get("/{user_id}", response_model=UserRead)
async def read_user(user_id: int, db: DBSession, current_user: CurrentUser):
result = await db.execute(select(User).where(User.id == user_id))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(status_code=404, detail="User not found")
return user
@router.delete("/me", status_code=status.HTTP_204_NO_CONTENT)
async def delete_current_user(current_user: CurrentUser, db: DBSession):
await db.delete(current_user)
await db.commit()
Items Router (Full CRUD)
# app/routers/items.py
from fastapi import APIRouter, HTTPException, status
from sqlalchemy import select
from app.dependencies import CurrentUser, DBSession
from app.models.item import Item
from app.schemas.item import ItemCreate, ItemRead, ItemUpdate
router = APIRouter(prefix="/items", tags=["items"])
@router.post("/", response_model=ItemRead, status_code=status.HTTP_201_CREATED)
async def create_item(item_in: ItemCreate, db: DBSession, current_user: CurrentUser):
item = Item(**item_in.model_dump(), owner_id=current_user.id)
db.add(item)
await db.commit()
await db.refresh(item)
return item
@router.get("/", response_model=list[ItemRead])
async def list_items(db: DBSession, current_user: CurrentUser, skip: int = 0, limit: int = 20):
result = await db.execute(
select(Item).where(Item.owner_id == current_user.id).offset(skip).limit(limit)
)
return result.scalars().all()
@router.get("/{item_id}", response_model=ItemRead)
async def read_item(item_id: int, db: DBSession, current_user: CurrentUser):
result = await db.execute(select(Item).where(Item.id == item_id, Item.owner_id == current_user.id))
item = result.scalar_one_or_none()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
return item
@router.patch("/{item_id}", response_model=ItemRead)
async def update_item(item_id: int, item_in: ItemUpdate, db: DBSession, current_user: CurrentUser):
result = await db.execute(select(Item).where(Item.id == item_id, Item.owner_id == current_user.id))
item = result.scalar_one_or_none()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
update_data = item_in.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(item, field, value)
await db.commit()
await db.refresh(item)
return item
@router.delete("/{item_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_item(item_id: int, db: DBSession, current_user: CurrentUser):
result = await db.execute(select(Item).where(Item.id == item_id, Item.owner_id == current_user.id))
item = result.scalar_one_or_none()
if not item:
raise HTTPException(status_code=404, detail="Item not found")
await db.delete(item)
await db.commit()
model_dump(exclude_unset=True) is a Pydantic v2 feature that only returns fields explicitly provided in the request body. This is essential for partial updates (PATCH): if the client sends only {"title": "new title"}, the description is not overwritten with None.
Application Factory
# app/main.py
from fastapi import FastAPI
from app.routers import auth, items, users
app = FastAPI(
title="Production API",
description="Users and Items REST API — FastAPI 2026 tutorial",
version="1.0.0",
)
app.include_router(auth.router)
app.include_router(users.router)
app.include_router(items.router)
@app.get("/health", tags=["health"])
async def health_check():
return {"status": "ok"}
Visiting http://localhost:8000/docs after startup gives you a fully interactive Swagger UI. Every endpoint, request body, query parameter, and response schema is documented automatically. Click "Authorize", paste your JWT token, and test protected routes directly in the browser — no Postman required.
Database Migrations with Alembic
Alembic manages schema migrations as versioned Python scripts. Set it up once and never manually run CREATE TABLE again.
alembic init alembic
Edit alembic/env.py to point at your models and use the async engine:
# alembic/env.py (relevant sections)
import asyncio
from logging.config import fileConfig
from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from app.database import Base
from app.models import user, item # noqa: F401 — import models so metadata is populated
from app.config import settings
config = context.config
config.set_main_option("sqlalchemy.url", settings.database_url)
fileConfig(config.config_file_name)
target_metadata = Base.metadata
def run_migrations_offline():
context.configure(url=settings.database_url, target_metadata=target_metadata, literal_binds=True)
with context.begin_transaction():
context.run_migrations()
def do_run_migrations(connection):
context.configure(connection=connection, target_metadata=target_metadata)
with context.begin_transaction():
context.run_migrations()
async def run_migrations_online():
connectable = async_engine_from_config(
config.get_section(config.config_ini_section),
prefix="sqlalchemy.",
)
async with connectable.connect() as connection:
await connection.run_sync(do_run_migrations)
await connectable.dispose()
if context.is_offline_mode():
run_migrations_offline()
else:
asyncio.run(run_migrations_online())
Generate and apply your first migration:
alembic revision --autogenerate -m "create users and items tables"
alembic upgrade head
Alembic compares Base.metadata against the live database schema and generates the diff as a migration script. Review it before applying — autogenerate is accurate but not infallible for complex constraints.
Docker Compose Setup
# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml
version: "3.9"
services:
db:
image: postgres:16-alpine
environment:
POSTGRES_USER: appuser
POSTGRES_PASSWORD: apppass
POSTGRES_DB: appdb
volumes:
- postgres_data:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U appuser -d appdb"]
interval: 5s
timeout: 5s
retries: 5
api:
build: .
ports:
- "8000:8000"
environment:
DATABASE_URL: postgresql+asyncpg://appuser:apppass@db:5432/appdb
SECRET_KEY: change-me-in-production
depends_on:
db:
condition: service_healthy
command: >
sh -c "alembic upgrade head && uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload"
volumes:
postgres_data:
Start the full stack:
docker compose up --build
The healthcheck on the db service ensures the API container waits for PostgreSQL to be ready before running migrations. The alembic upgrade head command runs migrations at container start — safe for development and staging; for production, consider a separate migration job in your CI/CD pipeline.
Testing with pytest and httpx
FastAPI applications are straightforward to test because httpx.AsyncClient can call the ASGI app directly in-process — no network overhead, no running server.
# tests/conftest.py
import asyncio
import pytest
import pytest_asyncio
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from app.database import Base, get_db
from app.main import app
TEST_DATABASE_URL = "postgresql+asyncpg://appuser:apppass@localhost:5432/testdb"
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
TestSessionLocal = async_sessionmaker(test_engine, expire_on_commit=False)
async def override_get_db():
async with TestSessionLocal() as session:
yield session
app.dependency_overrides[get_db] = override_get_db
@pytest_asyncio.fixture(scope="session", autouse=True)
async def setup_db():
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest_asyncio.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
yield ac
# tests/test_auth.py
import pytest
@pytest.mark.asyncio
async def test_register_and_login(client):
# Register
resp = await client.post("/auth/register", json={"email": "[email protected]", "password": "secret123"})
assert resp.status_code == 201
assert resp.json()["email"] == "[email protected]"
# Login
resp = await client.post(
"/auth/token",
data={"username": "[email protected]", "password": "secret123"},
)
assert resp.status_code == 200
assert "access_token" in resp.json()
@pytest.mark.asyncio
async def test_protected_route_without_token(client):
resp = await client.get("/users/me")
assert resp.status_code == 401
# tests/test_items.py
import pytest
async def get_token(client, email="[email protected]", password="pass1234"):
await client.post("/auth/register", json={"email": email, "password": password})
resp = await client.post("/auth/token", data={"username": email, "password": password})
return resp.json()["access_token"]
@pytest.mark.asyncio
async def test_create_and_list_items(client):
token = await get_token(client)
headers = {"Authorization": f"Bearer {token}"}
# Create
resp = await client.post("/items/", json={"title": "My Item", "description": "A test item"}, headers=headers)
assert resp.status_code == 201
item_id = resp.json()["id"]
# List
resp = await client.get("/items/", headers=headers)
assert resp.status_code == 200
assert any(i["id"] == item_id for i in resp.json())
@pytest.mark.asyncio
async def test_update_item(client):
token = await get_token(client, email="[email protected]")
headers = {"Authorization": f"Bearer {token}"}
resp = await client.post("/items/", json={"title": "Old Title"}, headers=headers)
item_id = resp.json()["id"]
resp = await client.patch(f"/items/{item_id}", json={"title": "New Title"}, headers=headers)
assert resp.status_code == 200
assert resp.json()["title"] == "New Title"
assert resp.json()["description"] is None # unchanged
Run all tests:
pytest tests/ -v --asyncio-mode=auto
The dependency_overrides mechanism in conftest.py is the clean way to swap get_db for a test database session without touching application code. All other dependencies that call get_db (including get_current_user) automatically use the override — the dependency tree resolves at request time.
OpenAPI / Swagger Docs
FastAPI generates OpenAPI 3.1 schema automatically. Two UIs ship out of the box:
- Swagger UI —
http://localhost:8000/docs— interactive, supports the "Authorize" button for Bearer tokens - ReDoc —
http://localhost:8000/redoc— clean read-only documentation, ideal for sharing with API consumers
The raw schema JSON is at http://localhost:8000/openapi.json. You can import this into Postman, generate client SDKs with openapi-generator, or validate it in CI with openapi-spec-validator.
To customize docs metadata, pass arguments to the FastAPI() constructor:
app = FastAPI(
title="My Production API",
description="Full description with **Markdown** support.",
version="1.0.0",
contact={"name": "API Support", "email": "[email protected]"},
license_info={"name": "MIT"},
)
To add response examples that appear in the docs, use openapi_examples in your Pydantic field or route decorator — no separate spec file needed.
Production Checklist
Before shipping to production, verify these items:
- Secret key: generate with
python -c "import secrets; print(secrets.token_hex(32))"and store in a secrets manager (AWS Secrets Manager, HashiCorp Vault, etc.), never in.envfiles committed to version control. - Database connection pool: set
pool_sizeandmax_overflowoncreate_async_engineto match your Postgresmax_connections. - CORS: add
CORSMiddlewareif your API is consumed by a browser frontend. - Rate limiting: use
slowapi(a FastAPI-compatible wrapper aroundlimits) for per-IP rate limiting. - Logging: configure structured JSON logging (via
structlogorpython-json-logger) so logs are parseable by Datadog, Loki, or CloudWatch. - Migrations in CI: run
alembic upgrade headas a pre-deploy step, not inside the application container startup, to avoid race conditions in multi-replica deployments. - Health checks: the
/healthendpoint should also verify the database connection (SELECT 1) before returning 200.
Summary
You now have a complete blueprint for a production FastAPI service:
- Async SQLAlchemy 2.0 with PostgreSQL via
asyncpg— the fastest Python database stack available - Pydantic v2 schemas for request validation and response serialization with
from_attributes=True - JWT authentication — password hashing with bcrypt, token creation and verification with
python-jose - Dependency injection — clean, reusable
Depends()callables for database sessions and the current user - Alembic migrations — autogenerated, versioned, async-compatible
- Docker Compose — one command to run the full stack with a healthy Postgres container
- pytest + httpx tests — in-process ASGI testing with dependency overrides
FastAPI's combination of speed, type safety, and zero-configuration documentation makes it the right choice for new Python API projects in 2026. The full project code structure above is the starting point — extend it with background tasks (BackgroundTasks), WebSocket endpoints, or a Celery worker queue as your application grows.