FastAPI · Advanced

Database

Talking to PostgreSQL without blocking: the engine, sessions, and create/read/update/delete.

SQLAlchemy

Overview

FastAPI does not ship a database layer; SQLAlchemy 2 with its async extension is the usual choice for relational databases. The engine owns a pool of connections, a session is a unit of work over one of them, and a yield dependency gives each request its own session that commits on success and rolls back on error.

Key concepts

Engine and pool: Create one engine per app. pool_size and max_overflow cap how many connections the app opens at once.
Session per request: A session must not be shared between concurrent requests, so each request gets its own from the dependency.
expire_on_commit=False: Keeps loaded attributes usable after commit - important in async code, where reloading them lazily is not allowed.
2.0 query style: Build queries with select(User).where(...), run them with await session.execute(...), then read results with .scalars() or .scalar_one_or_none().
Migrations: Alembic manages schema changes over time. Base.metadata.create_all is for prototypes only.

Best practices

Use an async driver (asyncpg, aiosqlite) with the async engine - a sync driver would block the event loop.
Load relationships explicitly with selectinload; implicit lazy loading raises errors in async sessions.
Keep queries in a repository or service layer rather than in route handlers.

Async SQLAlchemy + PostgreSQL

Engine, session, get_db dependency

Setup

# pip install sqlalchemy asyncpg (postgres) / aiosqlite (sqlite)
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/mydb"
# DEV: "sqlite+aiosqlite:///./dev.db"

engine = create_async_engine(
    DATABASE_URL, pool_size=10, max_overflow=20, pool_recycle=3600
)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

class Base(DeclarativeBase): pass

async def get_db():
    async with SessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

Watch out: Name the factory SessionLocal, not AsyncSession - that name belongs to SQLAlchemy's session class, and shadowing it breaks type hints like db: AsyncSession.

ORM model + CRUD operations

CRUD

from sqlalchemy import Column, Integer, String, Boolean, DateTime, select, update, delete
from sqlalchemy.sql import func

class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

# GET by PK
user = await db.get(User, user_id)

# GET by filter
result = await db.execute(select(User).where(User.email == email))
user = result.scalar_one_or_none()

# LIST with pagination + filter
result = await db.execute(
    select(User)
    .where(User.is_active == True)
    .order_by(User.created_at.desc())
    .offset(skip).limit(limit)
)
users = result.scalars().all()

# CREATE
user = User(email=data.email, hashed_password=hash_password(data.password))
db.add(user)
await db.flush()
await db.refresh(user)

# UPDATE
await db.execute(update(User).where(User.id == id).values(**data))

# DELETE
await db.execute(delete(User).where(User.id == id))

Comments

Loading comments...

Database

Overview

Key concepts

Best practices

Async SQLAlchemy + PostgreSQL

Engine, session, get_db dependency

ORM model + CRUD operations

Comments

AI

System Design

Backend

Frontend

Fundamentals