Chapter 4: the session is SQLAlchemy’s working memory. Identity map, unit of work, flush vs commit — getting these straight prevents most ORM frustration.

What a session is

A session is:

  • A connection (or pool of connections).
  • An identity map (for de-duplicating loaded rows).
  • A list of pending changes (the unit of work).
  • A transaction.

When you session.add(obj), you’re not writing to the DB yet — you’re queuing.

Creating sessions

from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

# Sync
SessionLocal = sessionmaker(engine, expire_on_commit=False)
with SessionLocal() as session:
    ...

# Async
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
async with AsyncSessionLocal() as session:
    ...

expire_on_commit=False is recommended for FastAPI / web apps. Why: by default, after commit(), all loaded objects are expired (next attribute access reloads from DB). For web requests where you commit and then return, that’s a wasted query.

Identity map

Within one session, the same primary key returns the same Python instance:

async with AsyncSessionLocal() as session:
    user1 = await session.get(User, 1)
    user2 = await session.get(User, 1)
    assert user1 is user2  # same object

This means session.add(existing_obj) is idempotent. And mutations on user1 are visible via user2.

The identity map is per-session; not shared across sessions.

Add, flush, commit

user = User(email="[email protected]")
session.add(user)        # queued; user.id is None
await session.flush()    # SQL INSERT runs; user.id is now set
await session.commit()   # transaction commits
  • add queues.
  • flush sends pending changes to the DB (still in transaction).
  • commit commits (also flushes if needed).

session.refresh(obj) re-reads attributes from DB after a flush.

Auto-flush

result = await session.execute(select(User).where(User.email == "x"))

Before SELECT, SQLAlchemy auto-flushes pending changes. So queries see your in-flight inserts/updates.

To disable per-session: with session.no_autoflush: ....

Commit semantics

commit ends the transaction. Any subsequent ORM op starts a new one.

For nested-like flow: savepoints / begin_nested().

async with session.begin_nested() as nested:
    session.add(obj)
    if oh_no:
        await nested.rollback()
        # outer transaction continues

Rollback

try:
    session.add(obj)
    await session.commit()
except IntegrityError:
    await session.rollback()

Rolls back; session is reusable.

After rollback, ORM-loaded objects may be stale; refresh if needed.

Expire and refresh

session.expire(user)     # marks all attrs as needing reload
session.expire(user, ["email"])  # specific attr
await session.refresh(user)  # immediately reload all

Useful when DB triggers update fields and you want fresh values.

The unit of work

Across a request:

async def transfer(session, from_id, to_id, amount):
    src = await session.get(Account, from_id, with_for_update=True)
    dst = await session.get(Account, to_id, with_for_update=True)
    src.balance -= amount
    dst.balance += amount
    await session.commit()

The session tracks dirty objects (mutations). On commit, it generates UPDATE for each. Order is decided by relationships and dependencies.

You don’t write UPDATE accounts SET ... — you mutate Python attributes; commit; SQLAlchemy figures out the SQL.

Persistent vs transient vs detached

  • Transient: not in any session. User(...) before add.
  • Pending: added but not flushed.
  • Persistent: in session, in DB.
  • Detached: was persistent; session closed or expunged.

Detached objects can be reattached: session.merge(obj).

Merge

detached_user = User(id=1, email="x")  # not in session
merged = await session.merge(detached_user)
# merged is now persistent; loaded current state from DB; copied dirty fields

For caching scenarios where you have an object outside a session and want to update.

For most CRUD: avoid merge. Fetch then mutate is clearer.

Closing

async with AsyncSessionLocal() as session: closes the session at exit.

Manual:

await session.close()

Closing rolls back any uncommitted transaction; returns connection to pool.

Per-request session pattern

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/users/{id}")
async def get_user(id: int, db: AsyncSession = Depends(get_db)):
    return await db.get(User, id)

Session per request. Auto-rolled-back on error; closed at the end.

For commit-on-success pattern:

async def get_db() -> AsyncSession:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

The handler doesn’t need to commit explicitly; the dep does on success.

I prefer explicit commits in the handler — it’s clearer when the boundary is. But auto-commit works fine; pick a convention and stick with it.

Multi-session per request

Rare but sometimes needed (e.g., write to one DB, read from another):

async def get_write_db(): ...
async def get_read_db(): ...

@app.post("/transfer")
async def transfer(write: AsyncSession = Depends(get_write_db), read: AsyncSession = Depends(get_read_db)):
    ...

scoped_session (sync only)

For sync code where you want a per-thread session without explicit passing:

from sqlalchemy.orm import scoped_session

ScopedSession = scoped_session(SessionLocal)

For async / FastAPI: use Depends instead. scoped_session is legacy.

flush() patterns

session.add(parent)
await session.flush()  # parent.id is set
child = Child(parent_id=parent.id)
session.add(child)
await session.commit()

When you need the auto-generated id before commit. Otherwise, relationships handle ordering automatically.

Event hooks

from sqlalchemy import event

@event.listens_for(Session, "before_commit")
def before_commit(session):
    # inspect session.new, session.dirty, session.deleted
    ...

Hooks let you audit / log every commit.

Sessions and threads / async

A session is not thread-safe. Don’t share across threads.

For async: a session is bound to one event loop. Don’t share across loops. Don’t use one session in concurrent tasks unless you serialize access.

Common mistakes

1. Long-lived session

session = AsyncSessionLocal()  # never closes

Hangs onto connections. Always context-manage.

2. Mixing sessions

user = await session1.get(User, 1)
session2.add(user)  # detached from session1; weirdness

One object, one session at a time.

3. Committing per-loop iteration

for u in users:
    session.add(u)
    await session.commit()  # 1000 commits

Batch:

session.add_all(users)
await session.commit()

4. Forgetting to refresh after commit (when expire_on_commit=True)

user.email after commit triggers re-fetch. Surprising perf hit.

Set expire_on_commit=False in async_sessionmaker.

5. Catching IntegrityError but not rolling back

Session is in a bad state; next op fails. Always rollback in except.

What’s next

Chapter 5: Querying with select().

Read this next

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 .