Scope SQLAlchemy 2.0 Sessions with Session.begin() for safe transactional writes

Use a short-lived Session per unit of work, not a global session

The common failure mode in SQLAlchemy 2.0 code is a long-lived Session that leaks connections, returns stale objects, and raises DetachedInstanceError when lazy attributes are accessed after close. The practical fix is to scope the Session to a single business transaction with Session.begin(), use 2.0 declarative mapping with Mapped and mapped_column, and execute queries with select().

This gives atomic commit or rollback, automatic identity map cleanup, and predictable attribute expiration.

Mechanism: declarative mapping and transactional Session

SQLAlchemy 2.0 declarative mapping maps a Python class to a table via a DeclarativeBase subclass. Attributes are typed with Mapped[T] and configured with mapped_column. The Session implements unit of work and identity map: changes are tracked, flushed to the database on commit, and the same primary key is loaded once per Session.

A minimal configuration for transactional persistence looks like this. Run in a Python 3.7+ interpreter with SQLAlchemy 2.0 installed. No special permissions are required for an in-memory SQLite engine; for a real database you need connect permission for the chosen user.

from sqlalchemy import String, Integer, create_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session, sessionmaker

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    name: Mapped[str] = mapped_column(String(100), nullable=False)

engine = create_engine("sqlite+pysqlite:///:memory:")
Base.metadata.create_all(engine)

SessionMaker = sessionmaker(bind=engine)

# write unit of work
with SessionMaker.begin() as s:
    s.add(User(name="Ada"))

# read unit of work
with SessionMaker() as s:
    user = s.get(User, 1)

Session.begin() opens a transaction on entry and commits on normal exit or rolls back on exception. The context manager guarantees the connection is returned to the pool. Using s.get or s.scalars(select(User).where(User.id == 1)).one() keeps queries in 2.0 style; session.query() is legacy in 2.0.

To verify mapping and persistence, enable SQL echo on the engine and inspect the generated statements for CREATE TABLE and INSERT. Opening a new Session for the read confirms the row was flushed and committed, not just held in the identity map of the writer Session.

Limits of the Session and ORM

Session is not thread safe. Do not share a Session instance across threads or async tasks. Each request or worker should create its own Session.

Identity map means the same primary key is returned as the same Python object within one Session. This masks concurrent updates from other connections until the Session is expired or refreshed.

expire_on_commit defaults to True. After commit, attributes are expired and the next access triggers a SELECT. Setting expire_on_commit=False keeps attribute values in memory but risks stale reads if another transaction modified the row.

SQL generation depends on the dialect. Features such as server-side defaults, RETURNING, and upsert syntax vary by database. Test against the target dialect.

Memory grows with long-lived Sessions because the identity map retains every loaded instance. Scope Sessions tightly.

Common mistakes and how to avoid them

• Opening a Session per query without closing. Always use a context manager or with Session.begin() so __exit__ returns the connection. Leaked connections exhaust the pool.
• Accessing lazy relationships after session close. Lazy loads require an active Session and an open connection. Access needed attributes before exiting the with block, or use joinedload or selectinload to eager load.
• N+1 queries in loops. Loading a collection with a loop triggers one SELECT per item. Replace with selectinload or joinedload at query time and verify with SQL echo that only one or two round trips occur.
• Missing Mapped annotations. In 2.0 style declarative, Mapped[T] is required for proper mapper configuration. Omitting it produces mapper errors at class definition time.

Check results practically by creating a small script with an in-memory engine, inserting in one Session.begin() block, then querying in a fresh Session and confirming the object loads. Intentionally access a lazy relationship after the Session is closed to observe DetachedInstanceError, confirming lifecycle requirements.