Cheatsheet for sessions. Long-form: Textbook Ch 4 .
Create sessionmaker
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
# Sync
SessionLocal = sessionmaker(engine, expire_on_commit=False)
# Async (recommended for FastAPI)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
Always expire_on_commit=False for web apps. Otherwise every attribute access after commit re-queries.
Basic patterns
# Sync
with SessionLocal() as session:
user = session.get(User, 1)
user.email = "[email protected]"
session.commit()
# Async
async with AsyncSessionLocal() as session:
user = await session.get(User, 1)
user.email = "[email protected]"
await session.commit()
add / flush / commit
session.add(user) # queued
await session.flush() # SQL INSERT runs; user.id set
await session.commit() # transaction commits
session.add_all([u1, u2, u3]) # batch
Rollback
try:
session.add(user)
await session.commit()
except IntegrityError:
await session.rollback()
Session is reusable after rollback.
get vs scalar vs execute
# By PK (uses identity map cache)
user = await session.get(User, 1)
# By predicate, exactly one
user = await session.scalar(select(User).where(User.email == "x"))
# Multiple
result = await session.execute(select(User))
users = result.scalars().all()
Identity map
Within a session, same PK = same object:
u1 = await session.get(User, 1)
u2 = await session.get(User, 1)
assert u1 is u2
Expire / refresh
session.expire(user) # mark all attrs stale
session.expire(user, ["email"]) # specific
await session.refresh(user) # reload from DB
Savepoints (nested transactions)
async with session.begin():
session.add(parent)
async with session.begin_nested() as sp:
session.add(child1)
if oh_no:
await sp.rollback() # rolls back only the savepoint
session.add(child2)
# parent + child2 committed; child1 not
merge (re-attach detached objects)
detached = User(id=1, email="x") # not in session
merged = await session.merge(detached)
# loads current state; copies dirty fields
Avoid in normal CRUD; fetch-then-mutate is clearer.
Per-request session (FastAPI)
async def get_db() -> AsyncSession:
async with AsyncSessionLocal() as session:
yield session
# Commit-on-success
async def get_db_commit() -> AsyncSession:
async with AsyncSessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
I prefer handler-explicit commits — easier to see boundaries.
Multi-session per request
async def transfer(write: AsyncSession = Depends(get_write_db),
read: AsyncSession = Depends(get_read_db)):
...
Autoflush
# Default: auto-flushes before SELECTs (so query sees uncommitted inserts)
# Disable for a block:
with session.no_autoflush:
...
Bulk via session
# Many INSERT in one statement
await session.execute(insert(User), [{"email": ...}, ...])
await session.commit()
For ORM events / relationships: still use session.add(user) per row.
Inspection
from sqlalchemy import inspect
state = inspect(user)
print(state.persistent, state.pending, state.detached, state.transient)
print(state.dict) # current attrs
print(state.committed_state) # before unflushed changes
Common mistakes
expire_on_commit=True(default) — extra query after commit.- One session across
asyncio.gathertasks — serialization / errors. - Long-held session during external HTTP — pool starvation.
- Catching
IntegrityErrorwithout rollback — session in bad state. - Mixing sessions — same object attached to two sessions.
Read this next
If you want my session-per-request + commit boundary patterns, it’s at rajpoot.dev .
Building something AI-, backend-, or data-heavy and want a second pair of eyes? I do consulting and freelance work — see my projects and ways to reach me at rajpoot.dev .