SQLAlchemy Cheat Sheet

01Install & Verifypip extras

pip install sqlalchemy★
Core + ORM. 2.0.52 is the current stable line.
pip install "sqlalchemy[asyncio]"2.1
greenlet no longer installs by default — async needs this extra.
pip install "sqlalchemy[postgresql]"
Driver extras: postgresql, mysql, oracle, mssql, aiosqlite…
pip install alembic★
Schema migrations — a separate package, same authors.
import sqlalchemy; sqlalchemy.__version__
Confirm which series you are actually on.

02Engine & Database URLsone per process

engine = create_engine(url)★
Lazy — no connection is opened until first use.
"sqlite+pysqlite:///file.db"
sqlite:// alone = in-memory, per-connection.
"postgresql+psycopg://u:p@host/db"2.1
Bare postgresql:// now defaults to psycopg 3, not psycopg2.
"mysql+pymysql://u:p@host/db"
MySQL / MariaDB. Also mysqldb, asyncmy.
"oracle+oracledb://…" · "mssql+pyodbc://…"2.1
Oracle default is now python-oracledb; new mssql+mssqlpython.
URL.create("postgresql", username=…, password=…)★
Builds the URL safely — no manual escaping of @ : /.
create_engine(url, echo=True)★
Log every statement to stdout. echo="debug" adds result rows.
engine.dispose()
Close the pool — call after fork(), never mid-request.

03Connections & TransactionsCore level

with engine.connect() as conn:
"Commit as you go" — rolls back on exit unless you commit.
with engine.begin() as conn:★
"Begin once" — commits on clean exit, rolls back on exception.
conn.execute(stmt, params)★
Returns a CursorResult. Pass a list of dicts for executemany.
conn.execute(text("SELECT * FROM t WHERE id=:i"), {"i": 1})
Raw SQL with bound params — never f-string user input.
conn.commit() · conn.rollback()
Explicit control inside engine.connect().
conn.begin_nested()
SAVEPOINT — a rollback point inside the outer transaction.
conn.execution_options(stream_results=True)
Server-side cursor for very large result sets.
conn.exec_driver_sql("…")
Straight to the DBAPI — no SQLAlchemy compilation at all.

04MetaData, Table, ColumnCore schema

md = MetaData()★
The registry of every Table you define. ORM: Base.metadata.
MetaData(schema="analytics")
Default schema for all tables — and, in 2.1, for named types too.
t = Table("user", md, Column("id", Integer, primary_key=True))★
Core table definition; the ORM builds one of these for you.
Column("name", String(50), nullable=False)
nullable defaults to True except on primary keys.
t.c.name · t.c["name"] · t.columns.keys()★
.c is the column collection you use in Core expressions.
t.primary_key · t.foreign_keys · md.sorted_tables
Introspect; sorted_tables is FK-dependency order.
class user_cols(TypedColumns): …2.1
Typed column collections so t.c.id infers as Column[int].

05Column Typesportable → dialect

Integer · BigInteger · SmallInteger
Python int.
String(50) · Text · Unicode★
Length is required by MySQL for indexed VARCHARs.
Numeric(10, 2) · Float · Double
Numeric → Decimal; use it for money.
Boolean · Date · DateTime(timezone=True) · Interval★
Set timezone=True or you get naive datetimes back.
Enum(MyPyEnum, name="kind")
Native ENUM where supported, else VARCHAR + CHECK.
JSON · ARRAY · Uuid
Uuid is the portable one; UUID is Postgres-native.
from sqlalchemy.dialects.postgresql import JSONB, INET, HSTORE
Dialect types unlock native operators and containment.
class MyType(TypeDecorator): impl = String
Custom bind/result coercion around an existing type.
registry(type_annotation_map={str: Text})
Map Python annotations → SQL types once, globally.

06Constraints, Indexes, DefaultsDDL detail

ForeignKey("user_account.id", ondelete="CASCADE")★
Goes on the many side. String form is resolved lazily.
UniqueConstraint("a", "b") · CheckConstraint("qty > 0")
Composite / expression constraints live in __table_args__.
Index("ix_user_name", User.name, unique=True)★
Or mapped_column(index=True) for the simple case.
__table_args__ = (UniqueConstraint("a","b"), {"schema": "s"})
A tuple; the dict of options must come last.
mapped_column(default=0)
Python-side default, computed at INSERT time by SQLAlchemy.
mapped_column(server_default=func.now())★
DDL-side DEFAULT — the database fills it in.
mapped_column(onupdate=func.now())
Touched on every UPDATE — the classic updated_at.
Computed("x * x", persisted=True)2.1
2.1 defaults to VIRTUAL on PG 18+; pass persisted=True for STORED.
Identity() · Sequence("user_id_seq")
Server-generated keys, modern and legacy spellings.

07Create, Drop & Reflectschema in & out

md.create_all(engine)★
CREATE IF NOT EXISTS semantics; never ALTERs an existing table.
md.drop_all(engine) · Table.create(engine)
Drops in reverse dependency order.
md.create_all(engine, checkfirst=CheckFirst.TABLES)2.1
Fine-grained control over which existence checks are emitted.
md.reflect(bind=engine)★
Load an existing database's schema into MetaData.
Table("user", md, autoload_with=engine)
Reflect just one table.
insp = inspect(engine); insp.get_table_names()★
Also get_columns, get_indexes, get_foreign_keys, has_table.
Base = automap_base(); Base.prepare(engine)
Generate mapped classes from a live schema.
CreateView(select(t), "v_users", metadata=md)2.1
First-class CREATE VIEW; also CreateTableAs and .into().
print(CreateTable(t).compile(engine))
See the DDL without running it.

08select() — the builderCore and ORM alike

select(User)★
All mapped columns; rows come back as User objects.
select(User.id, User.name)★
Named columns; rows are Row tuples, not objects.
.where(…).order_by(…).limit(10).offset(20)★
Every method returns a new statement — reassign it.
.order_by(User.name.desc(), User.id.asc())
Add .nulls_last() where the backend supports it.
.group_by(User.city).having(func.count() > 5)
HAVING filters groups; WHERE filters rows.
.distinct()
Postgres-only distinct_on() lives in the dialect module.
.with_only_columns(User.id)
Replace the column list, keeping the FROM/WHERE.
select(User).filter_by(name="ada")2.1
2.1 searches all FROM entities; ambiguity now raises.
print(stmt) · stmt.compile(engine)★
See the SQL. Add literal_binds to inline the values.
tstring(t"SELECT * FROM u WHERE id={uid}")2.1
PEP 750 t-strings: literals bind, SQL objects embed. Py 3.14+.

09WHERE & Operatorscolumns overload Python

.where(User.age >= 18, User.active)★
Multiple args are ANDed together.
and_(a, b) · or_(a, b) · not_(a)★
Needed for OR — Python's and/or do not overload.
User.name.in_(["a", "b"]) · .not_in(…)★
Accepts a subquery too.
User.bio.is_(None) · .is_not(None)★
Use these, not == None, for readable NULL tests.
User.name.like("a%") · .ilike(…) · .startswith(…)
Also contains, endswith, regexp_match.
User.age.between(18, 65)
Inclusive on both ends.
User.posts.any(Post.title == "x")★
EXISTS on a collection; .has() for many-to-one.
bindparam("x") · literal(5) · text("a = :x")
Explicit parameters when the operator isn't enough.
cast(col, String) · type_coerce(col, JSON)
SQL CAST vs. a Python-side type hint with no CAST emitted.
json_col.contains("v")deprecated
2.1: string-LIKE on JSON warns — use type_coerce(…, String) or JSONB.

10Joins, Subqueries, CTEsshaping the FROM

select(User).join(User.posts)★
ON clause inferred from the relationship.
.join(Post, Post.user_id == User.id)
Explicit ON when there's no relationship or it's ambiguous.
.outerjoin(Post) · .join(…, isouter=True)★
LEFT OUTER JOIN. full=True for FULL OUTER.
.join_from(User, Post)
State both sides when the left side isn't obvious.
p = aliased(Post); .join(p, …)★
Required to join the same table twice.
sq = select(…).subquery(); select(sq.c.id)
Anonymous derived table; address columns through .c.
select(…).scalar_subquery()
A one-column subquery usable inside an expression.
cte = select(…).cte("c", recursive=True)
WITH … ; grow it with cte.union_all(…).
union_all(s1, s2) · intersect(…) · except_(…)
Compound selects; wrap in .subquery() to filter further.
select(…).lateral() · .correlate(t)
LATERAL joins and explicit correlation control.

11Functions & Aggregatesthe func namespace

func.count() · func.count(User.id)★
func.<anything> renders as a SQL function call.
func.sum · func.avg · func.min · func.max
Return types are inferred from the argument.
func.now() · func.current_date()★
Server-side clock — ideal as a server_default.
func.coalesce(a, b) · func.lower(x) · func.concat(…)
Anything the backend knows, func can spell.
case((cond, "yes"), else_="no")
SQL CASE as a Python expression.
func.row_number().over(partition_by=…, order_by=…)★
Window functions on any aggregate or ranking function.
.over(range_=FrameClause(…))2.1
RANGE frames now accept dates/intervals, not just integers.
func.count().filter(User.active)
FILTER (WHERE …) where supported.
select(func.json_each(col).table_valued("value"))
Table-valued and set-returning functions.

12INSERT / UPDATE / DELETECore DML

insert(User).values(name="ada")★
Or omit .values() and pass params to execute().
conn.execute(insert(User), [{…}, {…}])★
Batched executemany — far faster than a loop.
insert(User).returning(User.id)★
RETURNING is used automatically for ORM inserts where supported.
insert(User).from_select(["name"], select(…))
INSERT … SELECT, no round trip through Python.
update(User).where(…).values(active=False)★
Set-based UPDATE — skips the ORM's per-object machinery.
update(User).values(hits=User.hits + 1)
Atomic in-place update using the column itself.
delete(User).where(User.active.is_(False))
Result's .rowcount tells you how many rows matched.
delete(u).using(u.outerjoin(a, …))2.1
Explicit USING for multi-table DELETE on PG / MySQL.
postgresql.insert(t).on_conflict_do_update(…)★
UPSERT. MySQL: on_duplicate_key_update; SQLite has its own.
session.execute(update(User)…)care
Bulk DML bypasses in-memory objects — set synchronize_session.

13Declarative Mappingclasses → tables

class Base(DeclarativeBase): pass★
The 2.0 base. Replaces declarative_base().
__tablename__ = "user_account"★
Required unless you attach a __table__ yourself.
id: Mapped[int] = mapped_column(primary_key=True)★
Every mapped class needs at least one primary key column.
name: Mapped[str]★
Annotation alone is enough: String, NOT NULL.
bio: Mapped[str | None]★
| None is what makes the column NULLable.
mapped_column("db_name", String(50), unique=True)
First positional arg overrides the DB column name.
mapped_column(deferred=True)
Skip a heavy column until it's touched.
reg = registry(); @reg.mapped_as_dataclass
Decorator style — map a class without inheriting a base.
class Base(MappedAsDataclass, DeclarativeBase)
Real dataclasses: generated __init__, __repr__, __eq__.
User.__table__ · inspect(User).mapper
Drop from ORM down to the Core Table / Mapper.

14relationship()Python-side links

posts: Mapped[list["Post"]] = relationship(back_populates="user")★
One-to-many. The FK lives on Post, not here.
user: Mapped["User"] = relationship(back_populates="posts")★
Many-to-one, the other half of the pair.
relationship(back_populates=lambda: Post.user)2.1
Callable / direct attribute refs — lintable, no magic strings.
relationship(secondary=assoc_table)★
Many-to-many through a plain association Table.
relationship("Child", cascade="all, delete-orphan")★
Delete children with the parent. One-to-many side only.
relationship(uselist=False)
Force one-to-one on the "one" side.
relationship(primaryjoin=…, foreign_keys=[…])
Disambiguate when two FKs point at the same table.
relationship(order_by=Post.created)
Sort the collection every time it loads.
relationship(lazy="selectin")★
Set the default load strategy at mapping time.
WriteOnlyMapped["Post"] · DynamicMapped["Post"]
Huge collections you never want fully loaded.
relationship(backref="user")legacy
Works, but back_populates on both sides is explicit and typed.

15Mapping Extrasmixins · inheritance

class TimestampMixin: created: Mapped[datetime]★
Plain mixin classes compose into any mapped class.
@declared_attr def __tablename__(cls): …
Compute table names or columns per subclass.
__mapper_args__ = {"polymorphic_on": …}
Single- and joined-table inheritance.
__abstract__ = True
A base class that is never mapped to a table itself.
@hybrid_property def full(self): …★
Works in Python and compiles into SQL in a WHERE clause.
@coords.inplace.bulk_dml2.1
Hybrids can now expand into ORM bulk INSERT/UPDATE dicts.
composite(mapped_column("x"), mapped_column("y"))
Two columns behind one value object. 2.1 adds return_none_on.
column_property(select(…).scalar_subquery())
A read-only derived column computed by the database.
@validates("email") def check(self, key, v): …
Python-side validation on attribute set.
association_proxy("tags", "name")
Flatten a many-to-many into a list of scalars.

16The Sessionshort-lived, not shared

with Session(engine) as s:★
One session per request / task. Not thread-safe.
SessionLocal = sessionmaker(engine, expire_on_commit=False)★
A configured factory — the usual app-level pattern.
with Session(engine) as s, s.begin():★
Begin-once: commit on clean exit, rollback on exception.
s.add(obj) · s.add_all([…])★
Marks pending — no SQL yet.
s.delete(obj)
Marks for DELETE at the next flush, cascades included.
s.flush()
Emit the pending SQL, stay inside the transaction.
s.commit() · s.rollback()★
commit = flush + COMMIT. After a failed flush you must rollback.
s.refresh(obj) · s.expire(obj)
Re-SELECT now, vs. mark stale and reload on next access.
s.begin_nested()
SAVEPOINT — partial rollback inside a bigger transaction.
Session(engine, execution_options={…})2.1
Options now apply to flushes and eager loaders too.
scoped_session(SessionLocal)
Thread-local registry; prefer explicit passing in new code.

17ORM Querying2.0 style

s.scalars(select(User)).all()★
The everyday call: a list of User objects.
s.execute(select(User.id, User.name)).all()★
A list of Row tuples when you select columns.
s.get(User, 1)★
PK lookup; checks the identity map first, so it may emit no SQL.
s.get(User, {"a": 1, "b": 2}) · s.get_one(…)
Composite keys; get_one raises instead of returning None.
s.scalar(select(func.count()).select_from(User))★
Counting rows — never len(s.scalars(…).all()).
s.scalars(stmt).unique().all()★
Required when joinedload pulls a collection.
stmt.options(…) · stmt.execution_options(…)
Loader options vs. engine-level options.
s.execute(stmt, execution_options={"yield_per": 1000})
Stream large ORM results in batches.
select(User).where(User.id.in_(select(…)))
ORM entities compose freely with Core constructs.
s.query(User).filter_by(name="x").first()legacy
The 1.x Query API still works but is no longer documented first.

18Reading ResultsResult navigation

result.all()★
List of Row. Iterating a Result yields Rows too.
result.scalars()★
Unwrap the first column — the objects, not 1-tuples.
.one() · .one_or_none() · .first()★
one() raises on 0 or >1; first() just takes the head.
.scalar_one() · .scalar_one_or_none()★
Scalars + strictness in one call.
result.mappings().all()
Dict-like rows — handy for JSON serialization.
row.name · row[0] · a, b = row
Row is a named tuple: index, attribute or unpack.
result.rowcount · result.inserted_primary_key
DML feedback on a CursorResult.
result.partitions(500) · result.freeze()
Chunked iteration; freeze to re-read a Result.
stmt: Select[int, str]2.1
PEP 646 drops the Tuple[] wrapper — needs mypy 1.7+.

19Loading Strategiesthe N+1 cure

.options(selectinload(User.posts))★
Best default for collections. One extra SELECT … WHERE id IN (…).
.options(joinedload(Post.user))★
Best for many-to-one: LEFT JOIN in the same statement.
.options(subqueryload(User.posts))
Older eager form; selectinload is usually faster.
.options(lazyload(User.posts))
The default: a SELECT fires the moment you touch the attribute.
.options(raiseload(User.posts))★
Turn accidental lazy loads into a loud error. Great in tests.
relationship(lazy="raise_on_sql")
Allow it if it's already loaded, raise if it would hit the DB.
.options(selectinload(User.posts).selectinload(Post.tags))
Chain to walk more than one level deep.
.join(User.posts).options(contains_eager(User.posts))★
Reuse a JOIN you wrote to populate the collection.
.options(load_only(User.id, User.name))
Column-level: also defer(), undefer().
with_loader_criteria(Post, Post.active)
Inject a WHERE into every eager load of that entity.

20Object States & Identitywhat the Session knows

inspect(obj).transient / pending / persistent / detached★
The four states every mapped instance moves through.
s.new · s.dirty · s.deleted · s.identity_map
Everything the unit of work is currently tracking.
obj in s · object_session(obj)
Which session, if any, owns this instance.
s.expunge(obj)
Detach without deleting — the object keeps its loaded values.
s.merge(obj)
Copy a detached object's state back into this session.
s.expire_all() · s.expunge_all()
Blanket versions of the above.
get_history(obj, "name")
Added / unchanged / deleted values for one attribute.
obj.attr # after the session closeddetached
Eager-load before leaving scope, or set expire_on_commit=False.

21Asynciosame API, awaited

engine = create_async_engine("postgresql+asyncpg://…")★
Needs an async driver: asyncpg, aiosqlite, asyncmy, psycopg.
Sess = async_sessionmaker(engine, expire_on_commit=False)★
Almost always set expire_on_commit=False under async.
async with Sess() as s: await s.execute(stmt)★
Same select() objects — only the I/O calls are awaited.
(await s.scalars(stmt)).all()★
Await the call, then use the Result synchronously.
await conn.run_sync(Base.metadata.create_all)
run_sync bridges any sync-only API into the loop.
class Base(AsyncAttrs, DeclarativeBase)★
Then await obj.awaitable_attrs.posts for a lazy load.
.options(selectinload(User.posts))★
Eager-load up front — the real fix, not awaitable_attrs.
async_scoped_session(Sess, scopefunc=current_task)
Task-scoped registry when you can't pass the session down.
obj.posts # plain access under asyncMissingGreenlet
A lazy load has no greenlet to suspend into. Eager-load it.

22Pooling & Performancethe boring wins

create_engine(url, pool_pre_ping=True)★
Cheap liveness check — cures "server closed the connection".
pool_size=5, max_overflow=10, pool_timeout=30★
Ceiling is pool_size + max_overflow per process.
pool_recycle=1800
Recycle before MySQL's wait_timeout kills the socket.
poolclass=NullPool
No pooling — for serverless / short-lived processes.
engine.pool.status()
Checked-out vs. checked-in counts, live.
insertmanyvalues
Batches INSERTs with RETURNING automatically — on by default.
func.uuidv7(monotonic=True)2.1
Server-side monotonic keys now work as batch-INSERT sentinels.
create_engine(url, query_cache_size=1200)
Compiled-statement cache; watch for churn from generated SQL.
.options(raiseload("*"))★
The fastest way to find every N+1 in a code path.

23Events & Debugginghooks and echoes

create_engine(url, echo=True, echo_pool="debug")★
Fastest possible answer to "what SQL did that emit?".
logging.getLogger("sqlalchemy.engine").setLevel("INFO")
Route the same output through your app's logging config.
@event.listens_for(engine, "before_cursor_execute")★
Time queries, tag them, or log slow statements.
@event.listens_for(Session, "before_flush")
Also after_flush, after_commit, do_orm_execute.
@event.listens_for(User, "before_insert")
Mapper-level hooks per class.
@event.listens_for(Base, "resolve_type_annotation")2.1
New RegistryEvents — program your own annotation rules.
stmt.compile(engine, compile_kwargs={"literal_binds": True})★
SQL with the values inlined — for reading, not for running.
exc.IntegrityError · NoResultFound · MultipleResultsFound
All DBAPI errors arrive wrapped as DBAPIError subclasses.

24Alembic Migrationsseparate package

alembic init migrations★
Then point target_metadata = Base.metadata in env.py.
alembic revision --autogenerate -m "add users"★
Diffs models against the live DB. Always read the output.
alembic upgrade head · alembic downgrade -1★
head = newest revision on this branch.
alembic current · alembic history --verbose
Where the database is vs. where the scripts go.
alembic stamp head
Mark as migrated without running anything — for existing DBs.
alembic merge -m "merge" rev1 rev2
Reconcile two branches that both claim head.
alembic upgrade head --sql
Offline mode: print the SQL for a DBA to review.
md.create_all() # in productioncare
Creates missing tables but never migrates existing ones.

25New in 2.12.1.0b3 · beta

pip install "sqlalchemy[asyncio]"★
greenlet left the default install. The #1 upgrade surprise.
"postgresql://" → psycopg 3★
Pin postgresql+psycopg2:// if you still need the old driver.
tstring(t"… {value}")
PEP 750 template strings compile to bound parameters.
CreateView · CreateTableAs · SelectBase.into()
Views and CTAS are finally first-class DDL.
Select[int, str]
PEP 646 typing — Row._t workarounds can go.
autoflush is now unconditional★
Even a raw text() execute flushes pending changes first.
filter_by() searches all FROM entities
Duplicated names raise AmbiguousColumnError — use filter().
MappedAsDataclass defaults use DONT_SET
Passing only related_id= no longer NULLs it out.
composite() returns an object, not None
Opt back in with return_none_on=, or annotate | None.
Enum / DOMAIN belong to MetaData
They inherit its schema; inherit_schema is deprecated.
Select.ext() + SyntaxExtension
Add QUALIFY, INTO OUTFILE, etc. without patching the compiler.
float literals CAST to DOUBLE
Was FLOAT; matters for third-party dialects.

26Legacy → Modernwhat old tutorials teach

session.query(User)legacy
→ select(User) + session.execute() / session.scalars().
query.get(1)legacy
→ session.get(User, 1).
Base = declarative_base()legacy
→ class Base(DeclarativeBase): pass.
id = Column(Integer, primary_key=True)legacy
→ id: Mapped[int] = mapped_column(primary_key=True).
select([col1, col2])removed
→ select(col1, col2) — no list.
engine.execute("SELECT 1")removed
Connectionless execution is gone → with engine.connect().
Session(autocommit=True)removed
→ explicit begin() / commit().
sessionmaker(class_=AsyncSession)dated
→ async_sessionmaker(engine).
ClauseElement.params()deprecated
2.1 → the faster ExecutableStatement.params() on statements.

sqlalchemy 2.0 · 2.1

Four pictures that prevent most SQLAlchemy bugs

1 · the four object states

2 · N+1, and the two ways out

3 · where the foreign key lives

4 · what comes back from execute()

Appendix — moving 1.4 → 2.0 → 2.1

✚ adopt

⚠ changed

✚ adopt

⚠ removed

✚ adopt

⚠ watch out

Worth memorizing