Python Database Patterns

SQLAlchemy 2.0 and database best practices.

SQLAlchemy 2.0 Basics

from sqlalchemy import create_engine, select from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session

class Base(DeclarativeBase): pass

class User(Base): tablename = "users"

id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(String(100))
email: Mapped[str] = mapped_column(String(255), unique=True)
is_active: Mapped[bool] = mapped_column(default=True)

Create engine and tables

engine = create_engine("postgresql://user:pass@localhost/db") Base.metadata.create_all(engine)

Query with 2.0 style

with Session(engine) as session: stmt = select(User).where(User.is_active == True) users = session.execute(stmt).scalars().all()

Async SQLAlchemy

from sqlalchemy.ext.asyncio import ( AsyncSession, async_sessionmaker, create_async_engine, ) from sqlalchemy import select

Async engine

engine = create_async_engine( "postgresql+asyncpg://user:pass@localhost/db", echo=False, pool_size=5, max_overflow=10, )

Session factory

async_session = async_sessionmaker(engine, expire_on_commit=False)

Usage

async with async_session() as session: result = await session.execute(select(User).where(User.id == 1)) user = result.scalar_one_or_none()

Model Relationships

from sqlalchemy import ForeignKey from sqlalchemy.orm import relationship, Mapped, mapped_column

class User(Base): tablename = "users"

id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]

# One-to-many
posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base): tablename = "posts"

id: Mapped[int] = mapped_column(primary_key=True)
title: Mapped[str]
author_id: Mapped[int] = mapped_column(ForeignKey("users.id"))

# Many-to-one
author: Mapped["User"] = relationship(back_populates="posts")

Common Query Patterns

from sqlalchemy import select, and_, or_, func

Basic select

stmt = select(User).where(User.is_active == True)

Multiple conditions

stmt = select(User).where( and_( User.is_active == True, User.age >= 18 ) )

OR conditions

stmt = select(User).where( or_(User.role == "admin", User.role == "moderator") )

Ordering and limiting

stmt = select(User).order_by(User.created_at.desc()).limit(10)

Aggregates

stmt = select(func.count(User.id)).where(User.is_active == True)

Joins

stmt = select(User, Post).join(Post, User.id == Post.author_id)

Eager loading

from sqlalchemy.orm import selectinload stmt = select(User).options(selectinload(User.posts))

FastAPI Integration

from fastapi import Depends, FastAPI from sqlalchemy.ext.asyncio import AsyncSession from typing import Annotated

async def get_db() -> AsyncGenerator[AsyncSession, None]: async with async_session() as session: yield session

DB = Annotated[AsyncSession, Depends(get_db)]

@app.get("/users/{user_id}") async def get_user(user_id: int, db: DB): 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) return user

Quick Reference

Operation SQLAlchemy 2.0 Style

Select all select(User)

Filter .where(User.id == 1)

First .scalar_one_or_none()

All .scalars().all()

Count select(func.count(User.id))

Join .join(Post)

Eager load .options(selectinload(User.posts))

Additional Resources

./references/sqlalchemy-async.md
Async patterns, session management
./references/connection-pooling.md
Pool configuration, health checks
./references/transactions.md
Transaction patterns, isolation levels
./references/migrations.md
Alembic setup, migration strategies

Assets

./assets/alembic.ini.template
Alembic configuration template

python-database-patterns

Safety Notice

Copy this and send it to your AI assistant to learn