SQLAlchemy Core, ORM & sessions

Core vs ORM — when do you use each?

Core is a SQL expression language: you compose select(), insert() and joins against Table objects and get rows back. ORM adds mapped classes, an identity map, a unit of work with change tracking, and relationship loading.

Use the ORM for domain logic where objects and their graphs matter; drop to Core for bulk operations, reporting queries, complex analytical SQL, and ETL — where instantiating objects is pure overhead. They interoperate: session.execute(select(...)) runs a Core statement in the ORM's session and transaction.

Explain the session, the identity map and the unit of work.

A Session is a transaction-scoped workspace. It holds an identity map: within one session, one primary key maps to exactly one Python object — so two queries for user 1 return the same object, and a change to it is visible everywhere.

The unit of work collects changes and flushes them as SQL in dependency order at the right moment:

with Session(engine) as session, session.begin():   # begin() commits on exit, rolls back on error
    user = session.get(User, 1)
    user.email = "new@example.com"      # no SQL yet — just marked dirty
    session.add(Order(user=user))       # pending
    orders = session.scalars(select(Order)).all()   # autoflush: pending changes flushed FIRST
# COMMIT here

Key behaviours: flush() emits SQL inside the transaction (so you get generated ids) but does not commit; commit() flushes then commits; autoflush means a query can trigger a flush of half-finished objects, which surprises people (a NOT NULL violation from an object you were still building); and after commit() all instances are expired by default, so touching an attribute issues a fresh SELECT.

What is DetachedInstanceError and how do you avoid it?

You closed the session, then touched an attribute that was expired or a relationship that was never loaded — there is no connection left to load it.

Fixes, best first: load what you need while the session is open (eager loading with selectinload/joinedload); return plain data (a dataclass or Pydantic model) from the data layer rather than live ORM objects; or set expire_on_commit=False if you only need already-loaded scalars afterwards. Keeping ORM objects alive past their session boundary is a design smell — the session is the transaction.

Compare the relationship loading strategies.

stmt = (select(Order)
        .options(joinedload(Order.customer),                    # many-to-one -> join
                 selectinload(Order.items).joinedload(Item.product),  # collection -> IN query
                 raiseload("*"))                                # anything else is a bug
        .where(Order.status == "paid"))
orders = session.scalars(stmt).unique().all()   # .unique() required with joinedload on collections

raiseload("*") in your hot endpoints is the trick that turns a silent N+1 into a loud test failure.

How do you do bulk inserts and updates efficiently?

session.execute(insert(User), [{"email": e} for e in emails])       # executemany, fast
session.execute(update(User).where(User.active == False).values(archived=True))
session.execute(insert(User).values(rows).on_conflict_do_update(     # Postgres upsert
    index_elements=["email"], set_={"name": text("EXCLUDED.name")}))

Adding 50,000 objects one at a time with session.add() is slow because the unit of work tracks every object. Bulk statements skip the ORM's per-object machinery — and therefore skip ORM events, cascades and Python-side defaults. That is the trade-off to state.

How do async sessions work, and what is the biggest gotcha?

engine = create_async_engine(URL, pool_size=10, max_overflow=20, pool_pre_ping=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

async with SessionLocal() as session:
    result = await session.scalars(
        select(Order).options(selectinload(Order.items)).where(Order.id == oid))
    order = result.one()

The gotcha: lazy loading does not work in async. Touching an unloaded relationship would need to emit I/O from a synchronous attribute access, so it raises MissingGreenlet. Every relationship you intend to use must be eagerly loaded in the query (or accessed via await session.refresh(obj, ["items"]) / AsyncAttrs). expire_on_commit=False is near-mandatory for the same reason.

Also: the async driver must be async (asyncpg, aiomysql), and a session is not safe to share across concurrent tasks — one session per task.

How do you configure the connection pool, and what goes wrong?

create_engine(URL, pool_size=5, max_overflow=10, pool_timeout=30,
              pool_recycle=1800, pool_pre_ping=True)

• pool_size + max_overflow per process × number of processes must stay under the database's max_connections. Four Gunicorn workers × (5+10) = 60 connections from one pod — multiply by pods and you will exhaust Postgres. This is a very common outage.
• pool_recycle under the database/proxy idle timeout, and pool_pre_ping to detect dead connections, prevent "server has gone away" after an idle period or a failover.
• NullPool when running behind PgBouncer in transaction mode (double pooling causes prepared-statement conflicts — with asyncpg you also need statement_cache_size=0).
• Watch for pool exhaustion symptoms: TimeoutError: QueuePool limit of size 5 overflow 10 reached almost always means a session is not being closed, or a long external call is happening while holding a connection.

Explain relationship configuration: back_populates, cascades and lazy defaults.

class Parent(Base):
    children: Mapped[list["Child"]] = relationship(
        back_populates="parent",
        cascade="all, delete-orphan",     # ORM-level: deleting a parent deletes children
        passive_deletes=True,             # let the DB's ON DELETE CASCADE do it instead
        lazy="raise",                     # fail loudly on accidental lazy load
    )

class Child(Base):
    parent_id: Mapped[int] = mapped_column(ForeignKey("parent.id", ondelete="CASCADE"))
    parent: Mapped[Parent] = relationship(back_populates="children")

back_populates keeps both sides in sync in Python (backref is the older implicit form — prefer the explicit one). delete-orphan means removing a child from the collection deletes it. passive_deletes=True with a database-level ON DELETE CASCADE avoids loading thousands of children into memory just to delete them — an important performance detail.

How does Alembic fit in, and what do you check in a generated migration?

Alembic autogenerates a migration by comparing your models to the database. Always read the generated file. It commonly misses: column type changes on some backends, server defaults, constraint and index renames, and anything in a schema it is not told to reflect (include_object/include_schemas). It never generates data migrations.

Production practice: one head at a time (merge branches deliberately), migrations must be backwards-compatible with the running code (expand/contract), create indexes concurrently on large tables (op.create_index(..., postgresql_concurrently=True) with autocommit_block()), batch data backfills, and test the downgrade path or explicitly declare that you do not support it.

How do you test code that uses SQLAlchemy?

Use a real database (a container via testcontainers, or a dedicated test schema) — SQLite behaves differently enough (types, constraints, concurrency, no RETURNING in older versions) that passing tests will hide real bugs. Wrap each test in a transaction and roll it back for speed and isolation:

@pytest.fixture
def session(engine):
    conn = engine.connect(); trans = conn.begin()
    s = Session(bind=conn, join_transaction_mode="create_savepoint")
    yield s
    s.close(); trans.rollback(); conn.close()

And assert on query counts for hot paths — sqlalchemy.event.listen(engine, "before_cursor_execute", …) to count statements — so an accidental N+1 fails the build rather than the pager.