|$ curl https://forge-ai.dev/api/markdown?path=docs/python/sqlalchemy
$cat docs/sqlalchemy-2.0.md
updated Today·22-28 min read·published
SQLAlchemy 2.0
Introduction
SQLAlchemy 2.0 unifies Core (SQL expression language) and ORM with a consistent execution model: Connection / Session + select().
ℹ
info
Install: pip install "sqlalchemy[asyncio]" alembic.
Engine & Core
engine.py
Python
| 1 | from sqlalchemy import create_engine, MetaData, Table, Column, Integer, String, select, insert |
| 2 | |
| 3 | engine = create_engine("sqlite+pysqlite:///:memory:", echo=True, future=True) |
| 4 | metadata = MetaData() |
| 5 | users = Table( |
| 6 | "users", |
| 7 | metadata, |
| 8 | Column("id", Integer, primary_key=True), |
| 9 | Column("name", String(100), nullable=False), |
| 10 | ) |
| 11 | metadata.create_all(engine) |
| 12 | |
| 13 | with engine.begin() as conn: |
| 14 | conn.execute(insert(users), [{"name": "Ada"}, {"name": "Grace"}]) |
| 15 | rows = conn.execute(select(users).where(users.c.name == "Ada")).all() |
| 16 | print(rows) |
ORM 2.0 Style
orm20.py
Python
| 1 | from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, Session |
| 2 | from sqlalchemy import String, ForeignKey, create_engine, select |
| 3 | |
| 4 | class Base(DeclarativeBase): |
| 5 | pass |
| 6 | |
| 7 | class Author(Base): |
| 8 | __tablename__ = "authors" |
| 9 | id: Mapped[int] = mapped_column(primary_key=True) |
| 10 | name: Mapped[str] = mapped_column(String(120)) |
| 11 | books: Mapped[list["Book"]] = relationship(back_populates="author") |
| 12 | |
| 13 | class Book(Base): |
| 14 | __tablename__ = "books" |
| 15 | id: Mapped[int] = mapped_column(primary_key=True) |
| 16 | title: Mapped[str] = mapped_column(String(200)) |
| 17 | author_id: Mapped[int] = mapped_column(ForeignKey("authors.id")) |
| 18 | author: Mapped[Author] = relationship(back_populates="books") |
| 19 | |
| 20 | engine = create_engine("sqlite+pysqlite:///:memory:", future=True) |
| 21 | Base.metadata.create_all(engine) |
| 22 | |
| 23 | with Session(engine) as session: |
| 24 | ada = Author(name="Ada", books=[Book(title="Notes")]) |
| 25 | session.add(ada) |
| 26 | session.commit() |
| 27 | |
| 28 | stmt = select(Book).where(Book.title == "Notes") |
| 29 | book = session.scalars(stmt).first() |
| 30 | print(book.title if book else None) |
Session Patterns
session.py
Python
| 1 | from contextlib import contextmanager |
| 2 | from sqlalchemy.orm import sessionmaker, Session |
| 3 | from sqlalchemy import create_engine |
| 4 | |
| 5 | engine = create_engine("sqlite+pysqlite:///app.db", future=True) |
| 6 | SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) |
| 7 | |
| 8 | @contextmanager |
| 9 | def session_scope(): |
| 10 | session: Session = SessionLocal() |
| 11 | try: |
| 12 | yield session |
| 13 | session.commit() |
| 14 | except Exception: |
| 15 | session.rollback() |
| 16 | raise |
| 17 | finally: |
| 18 | session.close() |
| 19 | |
| 20 | # with session_scope() as s: |
| 21 | # s.add(...) |
✓
best practice
One session per request/unit of work. Do not share sessions across threads.
Relationships & Loading
loading.py
Python
| 1 | from sqlalchemy.orm import selectinload, joinedload |
| 2 | from sqlalchemy import select |
| 3 | |
| 4 | # stmt = ( |
| 5 | # select(Author) |
| 6 | # .options(selectinload(Author.books)) |
| 7 | # .where(Author.name == "Ada") |
| 8 | # ) |
| 9 | # author = session.scalars(stmt).one() |
| 10 | # list(author.books) |
| 11 | # |
| 12 | # joinedload: single JOIN (good for many-to-one) |
| 13 | # selectinload: SELECT IN (good for collections) |
| Loader | Use when |
|---|---|
| selectinload | Collections (1-N) |
| joinedload | Many-to-one / one-to-one |
| lazy='raise' | Catch accidental lazy IO |
| noload | Explicitly skip |
select() Cookbook
queries.py
Python
| 1 | from sqlalchemy import select, func, and_, or_ |
| 2 | |
| 3 | # select(Book).where(Book.title.ilike("%py%")) |
| 4 | # select(Author).where(Author.id.in_([1, 2, 3])) |
| 5 | # select(func.count()).select_from(Book).where(Book.author_id == 1) |
| 6 | # select(Book).order_by(Book.title.desc()).limit(10).offset(20) |
| 7 | # select(Author).where(or_(Author.name == "Ada", Author.name == "Grace")) |
Alembic Migrations
alembic.sh
Bash
| 1 | pip install alembic |
| 2 | alembic init migrations |
| 3 | # set sqlalchemy.url in alembic.ini or env.py |
| 4 | alembic revision --autogenerate -m "create authors books" |
| 5 | alembic upgrade head |
| 6 | alembic downgrade -1 |
Keep models and migrations in sync. Review autogenerate diffs — they are not perfect.
Async SQLAlchemy
async_sa.py
Python
| 1 | # from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession |
| 2 | # from sqlalchemy.orm import sessionmaker |
| 3 | # |
| 4 | # engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") |
| 5 | # AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) |
| 6 | # |
| 7 | # async with AsyncSessionLocal() as session: |
| 8 | # result = await session.scalars(select(User)) |
| 9 | # users = result.all() |
SQLAlchemy vs Django ORM
| Aspect | SQLAlchemy | Django ORM |
|---|---|---|
| Style | Explicit Session/Core | Manager on Model |
| Framework coupling | None | Django |
| Migrations | Alembic | Built-in |
| Async | First-class 2.0 | Improving |
| Best with | FastAPI / any | Django apps |
Transactions
tx.py
Python
| 1 | # with Session(engine) as session: |
| 2 | # with session.begin(): |
| 3 | # session.add(Author(name="Ada")) |
| 4 | # # commits on clean exit; rolls back on error |
| 5 | # |
| 6 | # nested: |
| 7 | # with session.begin_nested(): # SAVEPOINT |
| 8 | # ... |
Hybrid Properties (brief)
hybrid.py
Python
| 1 | from sqlalchemy.ext.hybrid import hybrid_property |
| 2 | from sqlalchemy.orm import Mapped, mapped_column, DeclarativeBase |
| 3 | from sqlalchemy import String |
| 4 | |
| 5 | class Base(DeclarativeBase): |
| 6 | pass |
| 7 | |
| 8 | class Person(Base): |
| 9 | __tablename__ = "people" |
| 10 | id: Mapped[int] = mapped_column(primary_key=True) |
| 11 | first: Mapped[str] = mapped_column(String(50)) |
| 12 | last: Mapped[str] = mapped_column(String(50)) |
| 13 | |
| 14 | @hybrid_property |
| 15 | def full_name(self) -> str: |
| 16 | return f"{self.first} {self.last}" |
Pitfalls
- DetachedInstanceError: access expired attrs outside session
- N+1 queries: use eager loading options
- autoflush surprises: know when flush happens
- Mixing Core and ORM objects without care
- Creating engines repeatedly — create once per process
Insert / Update / Delete
iud.py
Python
| 1 | from sqlalchemy import insert, update, delete |
| 2 | # conn.execute(insert(users).values(name="Ada")) |
| 3 | # conn.execute(update(users).where(users.c.id == 1).values(name="Grace")) |
| 4 | # conn.execute(delete(users).where(users.c.id == 1)) |
| 5 | # |
| 6 | # ORM: |
| 7 | # session.add(User(name="Ada")) |
| 8 | # session.execute(update(User).where(User.id == 1).values(name="Grace")) |
| 9 | # session.execute(delete(User).where(User.id == 1)) |
Engine URL Cheatsheet
| Backend | URL |
|---|---|
| SQLite | sqlite+pysqlite:///./app.db |
| Postgres | postgresql+psycopg://user:pass@host/db |
| Postgres async | postgresql+asyncpg://user:pass@host/db |
| MySQL | mysql+pymysql://user:pass@host/db |
Reflection (brief)
reflect.py
Python
| 1 | from sqlalchemy import create_engine, MetaData, Table |
| 2 | # engine = create_engine("sqlite:///app.db") |
| 3 | # meta = MetaData() |
| 4 | # users = Table("users", meta, autoload_with=engine) |
| 5 | # print(users.columns.keys()) |
$Blueprint — Engineering Documentation·Section ID: PYTHON-SQLA·Revision: 1.0
Community
Get help on Slack, Discord or VIP
Stuck on a guide? Join the community and ask.