FastAPI: Build High-Performance Python APIs in Record Time

pip install "fastapi[standard]"

from fastapi import FastAPI

app = FastAPI(
    title="My Awesome API",
    description="A production-ready API built with FastAPI",
    version="1.0.0",
)


# A simple health check endpoint
@app.get("/")
def root():
    return {"status": "healthy", "message": "Welcome to the API"}


@app.get("/items")
def list_items():
    return [
        {"id": 1, "name": "Laptop", "price": 999.99},
        {"id": 2, "name": "Keyboard", "price": 79.99},
    ]

uvicorn main:app --reload

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

app = FastAPI()


# Path parameter -- the item_id is part of the URL
@app.get("/items/{item_id}")
def get_item(
    item_id: int = Path(..., ge=1, description="The ID of the item to retrieve"),
):
    return {"item_id": item_id, "name": f"Item {item_id}"}


# Query parameters -- passed as ?skip=0&limit=10 in the URL
@app.get("/products")
def list_products(
    skip: int = Query(default=0, ge=0, description="Number of items to skip"),
    limit: int = Query(default=10, le=100, description="Max items to return"),
    category: Optional[str] = Query(default=None, description="Filter by category"),
):
    return {"skip": skip, "limit": limit, "category": category}


# Request body -- parsed from the JSON payload
class ItemCreate(BaseModel):
    name: str
    price: float
    description: Optional[str] = None
    in_stock: bool = True


@app.post("/items", status_code=201)
def create_item(item: ItemCreate):
    return {"message": "Item created", "item": item.model_dump()}

from pydantic import BaseModel, Field, EmailStr, field_validator
from datetime import datetime
from typing import Optional
from enum import Enum


class UserRole(str, Enum):
    admin = "admin"
    editor = "editor"
    viewer = "viewer"


# Base model with shared fields
class UserBase(BaseModel):
    email: EmailStr
    username: str = Field(
        ..., min_length=3, max_length=50,
        pattern=r"^[a-zA-Z0-9_]+$",
        description="Alphanumeric username, 3-50 characters",
    )
    full_name: Optional[str] = Field(default=None, max_length=100)
    role: UserRole = UserRole.viewer


# Model for creating a user (includes password)
class UserCreate(UserBase):
    password: str = Field(..., min_length=8, max_length=128)

    @field_validator("password")
    @classmethod
    def validate_password_strength(cls, v):
        if not any(c.isupper() for c in v):
            raise ValueError("Password must contain an uppercase letter")
        if not any(c.isdigit() for c in v):
            raise ValueError("Password must contain a digit")
        return v


# Model for responses (never expose password)
class UserResponse(UserBase):
    id: int
    is_active: bool = True
    created_at: datetime

    model_config = {"from_attributes": True}

app/
├── main.py           # creates FastAPI(lifespan=lifespan), includes routers
├── database.py       # engine, session maker, User model, lifespan
├── schemas.py        # Pydantic models
├── dependencies.py   # get_db, PaginationParams
├── users.py          # GET /users
└── auth.py           # POST /users, POST /auth/token, GET /users/me

from dataclasses import dataclass

from fastapi import Query

from database import async_session_maker


# A reusable pagination dependency
@dataclass
class PaginationParams:
    skip: int = Query(default=0, ge=0)
    limit: int = Query(default=20, ge=1, le=100)


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

from fastapi import APIRouter, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from database import User
from dependencies import PaginationParams, get_db
from schemas import UserResponse

router = APIRouter()


# Now use both dependencies in an endpoint
@router.get("/users", response_model=list[UserResponse])
async def list_users(
    pagination: PaginationParams = Depends(),
    db: AsyncSession = Depends(get_db),
):
    query = select(User).offset(pagination.skip).limit(pagination.limit)
    result = await db.execute(query)
    return result.scalars().all()

pip install "sqlalchemy[asyncio]" asyncpg

from contextlib import asynccontextmanager
from datetime import datetime, timezone

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy import String, Boolean, DateTime

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"

# Connection pooling is critical for production performance.
# pool_size controls steady-state connections; max_overflow
# allows burst capacity beyond that limit.
engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,          # keep 20 connections ready
    max_overflow=10,       # allow 10 extra under load
    pool_pre_ping=True,    # verify connections are alive
    pool_recycle=3600,     # recycle connections after 1 hour
    echo=False,
)

async_session_maker = async_sessionmaker(engine, expire_on_commit=False)


class Base(DeclarativeBase):
    pass


def utc_now() -> datetime:
    # datetime.utcnow() is deprecated since Python 3.12
    return datetime.now(timezone.utc)


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(String(255), unique=True, index=True)
    username: Mapped[str] = mapped_column(String(50), unique=True, index=True)
    hashed_password: Mapped[str] = mapped_column(String(255))
    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=utc_now
    )


# Application lifespan -- create tables on startup.
# Pass it to FastAPI(lifespan=lifespan) in main.py.
@asynccontextmanager
async def lifespan(app):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield
    await engine.dispose()

pip install pyjwt "pwdlib[argon2]"

from datetime import datetime, timedelta, timezone
from typing import Annotated

import jwt
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jwt.exceptions import InvalidTokenError
from pwdlib import PasswordHash
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from database import User
from dependencies import get_db
from schemas import UserCreate, UserResponse

# Generate with: openssl rand -hex 32 (load from an env variable in production)
SECRET_KEY = "change-me-to-a-long-random-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

password_hash = PasswordHash.recommended()  # Argon2
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/token")
router = APIRouter()


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


async def get_current_user(
    token: Annotated[str, Depends(oauth2_scheme)],
    db: AsyncSession = Depends(get_db),
):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        sub = payload.get("sub")
        if sub is None:
            raise credentials_exception
        user_id = int(sub)  # "sub" is a string in the token
    except (InvalidTokenError, ValueError):
        raise credentials_exception

    user = await db.get(User, user_id)
    if user is None:
        raise credentials_exception
    return user


# Sign-up endpoint -- store only the password hash
@router.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user_in: UserCreate, db: AsyncSession = Depends(get_db)):
    existing = await db.execute(
        select(User).where(
            (User.username == user_in.username) | (User.email == user_in.email)
        )
    )
    if existing.scalar_one_or_none() is not None:
        raise HTTPException(status_code=409, detail="Username or email already taken")

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


# Login endpoint
@router.post("/auth/token")
async def login(
    form_data: Annotated[OAuth2PasswordRequestForm, Depends()],
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(User).where(User.username == form_data.username)
    )
    user = result.scalar_one_or_none()
    if not user or not password_hash.verify(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Invalid credentials")

    access_token = create_access_token(
        data={"sub": str(user.id)},  # JWT "sub" must be a string
        expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES),
    )
    return {"access_token": access_token, "token_type": "bearer"}


# Protected endpoint -- requires valid JWT
@router.get("/users/me", response_model=UserResponse)
async def read_current_user(
    current_user: Annotated[User, Depends(get_current_user)],
):
    return current_user

from fastapi import FastAPI

import auth
import users
from database import lifespan

app = FastAPI(title="My Awesome API", lifespan=lifespan)
app.include_router(auth.router, tags=["auth"])
app.include_router(users.router, tags=["users"])