REF / WRITING · SOFTWARE

Python FastAPI: The Complete Guide to Building Production APIs

Learn FastAPI from zero to production - async endpoints, Pydantic v2 validation, SQLAlchemy, JWT auth, background tasks, and deployment. The complete reference.

DomainSoftware
Formattutorial
Published3 Sept 2024
Tagspython · fastapi · rest-api

FastAPI is the fastest-growing Python API framework. and for good reason. It combines Python type hints with automatic OpenAPI documentation, Pydantic v2 validation, and genuine async support. Teams routinely see 2-3× the throughput of Flask/Django for I/O-bound services, with less code and better developer ergonomics.

This guide is the complete reference: project setup, async database access, Pydantic v2 schemas, JWT authentication, background tasks, testing, and deployment. No shortcuts.

Why FastAPI in 2026?

Python has had web frameworks since the early 2000s. Flask and Django still have massive install bases. So why choose FastAPI?

  • Speed: performance comparable to Node.js and Go for async I/O workloads (Starlette + uvicorn underneath)
  • Automatic docs: OpenAPI/Swagger UI and ReDoc generated from your code, zero configuration
  • Pydantic v2: the validation layer is compiled in Rust; validation is 5-50× faster than v1
  • Type-driven development: your IDE and type checker catch bugs before runtime
  • Async-first: async def and await everywhere; no bolted-on async like older frameworks
  • Dependency injection: clean, testable DI with Depends()

Django is better when you need a batteries-included monolith (admin, ORM, auth all in one). Flask is fine for simple scripts promoted to APIs. FastAPI is the right choice when you're building a purpose-built API with performance requirements and a team that values type safety.

Prerequisites

  • Python 3.12+
  • uv or pip for package management
  • Familiarity with Python type hints
  • Basic understanding of HTTP and REST

Step 1: Project Setup

# Using uv (recommended - it's faster)
uv init myapi
cd myapi
uv add fastapi[standard] sqlalchemy asyncpg alembic pyjwt passlib[bcrypt]
uv add --dev pytest pytest-asyncio httpx

Or with pip:

pip install "fastapi[standard]" sqlalchemy asyncpg alembic pyjwt "passlib[bcrypt]"
pip install --dev pytest pytest-asyncio httpx

fastapi[standard] includes uvicorn, python-multipart, and email-validator: the essentials.

Step 2: Project Structure

myapi/
├── main.py                  # App factory + router registration
├── config.py                # Settings from environment
├── database.py              # Async SQLAlchemy engine + session
├── models/                  # SQLAlchemy ORM models
│   └── user.py
├── schemas/                 # Pydantic v2 request/response models
│   └── user.py
├── routers/                 # FastAPI APIRouter instances
│   ├── auth.py
│   └── users.py
├── services/                # Business logic
│   └── user_service.py
├── dependencies.py          # Reusable Depends() functions
└── alembic/                 # Database migrations

Step 3: Configuration

Use Pydantic Settings for environment-variable-backed config:

# config.py
from pydantic_settings import BaseSettings

class Settings(BaseSettings):
    database_url: str
    secret_key: str
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    debug: bool = False

    class Config:
        env_file = ".env"

settings = Settings()

.env file (never commit this):

DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/myapi
SECRET_KEY=your-very-long-random-secret-key

Step 4: Async Database Setup

# database.py
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase
from 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

The get_db function is a FastAPI dependency; Depends(get_db) injects a session into any route handler.

Step 5: Define the ORM Model

# models/user.py
from datetime import datetime, timezone
from sqlalchemy import String, Boolean, DateTime
from sqlalchemy.orm import Mapped, mapped_column
from database import Base

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, nullable=False, index=True)
    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),
        default=lambda: datetime.now(timezone.utc)
    )

SQLAlchemy 2.0's Mapped + mapped_column syntax is the modern approach. fully type-annotated, no more ambiguous Column() calls.

Step 6: Pydantic v2 Schemas

# schemas/user.py
from datetime import datetime
from pydantic import BaseModel, EmailStr, ConfigDict

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

class UserResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)  # replaces orm_mode=True

    id: int
    email: str
    is_active: bool
    created_at: datetime

class Token(BaseModel):
    access_token: str
    token_type: str = "bearer"

from_attributes=True (Pydantic v2) replaces the old orm_mode = True. It enables reading attributes from ORM model instances.

Step 7: Service Layer

# services/user_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from passlib.context import CryptContext
from models.user import User
from schemas.user import UserCreate

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

async def create_user(db: AsyncSession, user_in: UserCreate) -> User:
    # Check for duplicate email
    existing = await db.scalar(select(User).where(User.email == user_in.email))
    if existing:
        raise ValueError("Email already registered")

    user = User(
        email=user_in.email,
        hashed_password=pwd_context.hash(user_in.password),
    )
    db.add(user)
    await db.commit()
    await db.refresh(user)
    return user

async def authenticate_user(db: AsyncSession, email: str, password: str) -> User | None:
    user = await db.scalar(select(User).where(User.email == email))
    if not user or not pwd_context.verify(password, user.hashed_password):
        return None
    return user

Step 8: JWT Authentication

# dependencies.py
from datetime import datetime, timedelta, timezone
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from config import settings
from database import get_db
from models.user import User

security = HTTPBearer()

def create_access_token(user_id: int) -> str:
    expire = datetime.now(timezone.utc) + timedelta(minutes=settings.access_token_expire_minutes)
    return jwt.encode(
        {"sub": str(user_id), "exp": expire},
        settings.secret_key,
        algorithm=settings.algorithm,
    )

async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: AsyncSession = Depends(get_db),
) -> User:
    token = credentials.credentials
    try:
        payload = jwt.decode(token, settings.secret_key, algorithms=[settings.algorithm])
        user_id = int(payload["sub"])
    except (jwt.InvalidTokenError, KeyError, ValueError):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

    user = await db.get(User, user_id)
    if not user or not user.is_active:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Inactive user")
    return user

Step 9: Routers

# routers/auth.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from database import get_db
from schemas.user import UserCreate, UserResponse, Token
from services.user_service import create_user, authenticate_user
from dependencies import create_access_token

router = APIRouter(prefix="/auth", tags=["auth"])

@router.post("/register", response_model=UserResponse, status_code=201)
async def register(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    try:
        return await create_user(db, user_in)
    except ValueError as e:
        raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e))

@router.post("/login", response_model=Token)
async def login(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    user = await authenticate_user(db, user_in.email, user_in.password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Incorrect email or password",
        )
    return Token(access_token=create_access_token(user.id))

Step 10: App Factory

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from routers import auth, users

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Startup: initialize DB, warm caches, etc.
    yield
    # Shutdown: close connections, flush buffers, etc.

app = FastAPI(
    title="My API",
    version="1.0.0",
    lifespan=lifespan,
    docs_url="/docs" if settings.debug else None,  # Hide docs in production
)

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

app.include_router(auth.router, prefix="/api/v1")
app.include_router(users.router, prefix="/api/v1")

Step 11: Testing

# tests/test_auth.py
import pytest
from httpx import AsyncClient, ASGITransport
from main import app

@pytest.mark.asyncio
async def test_register_success():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        resp = await client.post("/api/v1/auth/register", json={
            "email": "test@example.com",
            "password": "strongpassword123"
        })
    assert resp.status_code == 201
    assert resp.json()["email"] == "test@example.com"

@pytest.mark.asyncio
async def test_login_wrong_password():
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client:
        resp = await client.post("/api/v1/auth/login", json={
            "email": "test@example.com",
            "password": "wrongpassword"
        })
    assert resp.status_code == 401

Common Pitfalls

1. Blocking the event loop: calling synchronous I/O (blocking DB drivers, requests library) inside async def blocks the entire event loop. Always use async drivers (asyncpg, httpx).

2. Session scope: create a new AsyncSession per request, not a module-level singleton. The get_db dependency pattern ensures this.

3. expire_on_commit=False: essential for async sessions. Without it, accessing ORM attributes after commit() triggers lazy loads that fail in async context.

4. Using Pydantic v1 syntax in v2: orm_mode = True is from_attributes = True in v2. Check the migration guide when upgrading.

5. Exposing /docs in production: Swagger UI leaks your API structure. Set docs_url=None in production unless you have auth on the docs endpoint.

When FastAPI vs Django vs Flask

NeedBest Choice
Async, high-throughput APIFastAPI
Admin panel + ORM + auth out of boxDjango
Simple script promoted to HTTP endpointFlask
ML model servingFastAPI (async, fast serialisation)
CMS / content siteDjango

FAQ

Q: Is FastAPI production-ready? Yes. Used by Netflix, Uber, Microsoft, and thousands of production teams. The underlying Starlette framework is mature and battle-tested.

Q: Do I need to use async everywhere? No. def (sync) route handlers are run in a threadpool automatically by FastAPI. Use async def when you have async I/O; use def for CPU-bound work or when using blocking libraries.

Q: How do I handle file uploads? Use UploadFile from fastapi. For large files, stream to object storage (S3, R2) rather than loading into memory.

Q: What about WebSockets? FastAPI has native WebSocket support: @app.websocket("/ws"). For the LLM streaming deep-dive, see the companion article.

Q: How does FastAPI compare to Flask performance-wise? For sync endpoints: similar. For async I/O workloads (most APIs): FastAPI is 3-10× faster depending on the benchmark. The difference compounds at high concurrency.

Conclusion

FastAPI is the modern Python API framework. Pydantic v2 validation, genuine async support, automatic OpenAPI docs, and clean dependency injection make it the right choice for production API development.

The patterns in this guide. async SQLAlchemy, Pydantic v2 schemas, JWT with HTTPBearer, layered service architecture. are the patterns production teams use.

Next: Streaming LLM Responses with FastAPI. how to stream real-time AI completions to clients with SSE and WebSockets.