SQLAlchemy resource and transaction ownership
| Object | Owns | Rule |
|---|---|---|
Engine |
Dialect plus connection pool | Create at application scope; it is not an active transaction. |
Connection |
Checked-out DBAPI connection and Core transaction state | Scope with a context manager; commit or roll back explicitly. |
Session |
ORM identity map and one logical transaction | One unit of work; never a global cache or cross-task singleton. |
Result |
Cursor-backed rows until consumed/closed | Consume inside the owning connection/session scope unless materialized. |
Workflow
- Inspect SQLAlchemy, driver, dialect, sync/async mode, transaction boundary, expected concurrency, isolation, and data volume.
- Choose Core for explicit relational statements and bulk/data-oriented work;
choose ORM when identity, relationships, and unit-of-work behavior are part
of the domain model. Both use 2.x
select()and transaction semantics. - Create Engine/AsyncEngine once per configured database role. Inject a connection or session factory; create one Session/AsyncSession per request, job, or unit of work.
- Use
with engine.begin() as connection:orwith Session.begin() as session:when the block should commit on success and roll back on error. Do not confuseflush()with commit. - Compose values with SQLAlchemy expressions and bound parameters. Never use string interpolation for untrusted values or dynamic identifiers.
- Define relationship loading deliberately. Detect N+1 access and detached lazy loads; use joined/select-in loading according to cardinality and result multiplication.
- Test rollback, uniqueness/integrity errors, transaction visibility, generated SQL/parameters when relevant, and backend-specific behavior.
Concurrency and async
Session and AsyncSession are mutable transactional objects. Do not share one
instance across threads or concurrent tasks. AsyncSession is a proxy over the
same stateful session model; use one per task and an async driver. Do not mix
sync driver calls into the event loop.
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session
engine = create_engine(url)
def load_user(user_id: int) -> User | None:
with Session(engine) as session:
return session.scalar(select(User).where(User.id == user_id))
If the returned mapped object will be used after Session close, ensure required attributes are loaded and understand expiration/detachment. Read Core and ORM decisions, transactions and concurrency, and query verification.