ORM Patterns That Don't Blow Up in Production
1 · The lesson
readThe ORM is a tool with sharp edges. Used well, it shaves weeks off a project and keeps the model code readable as the schema grows. Used badly, it generates the SQL equivalent of nightmare fuel — N+1 queries, full table scans on every page, transaction-less updates, and migrations that lock the database for fifteen minutes mid-deploy.
This lesson is the production playbook — the patterns and anti-patterns that decide whether your ORM-backed service stays performant at 10× your current load or buckles the day the marketing team gets lucky. Examples target SQLAlchemy 2.0 because it's the canonical Python ORM and the patterns translate to Django ORM and Tortoise ORM with minor syntax changes.
Examples assume a running local Postgres — Docker is the easiest way; see devops-docker. Pip-install the driver as shown. Expected output is in comments.
1. The ORM Trade-Off — Stated Honestly
An ORM trades control over SQL for speed of development. You write session.scalars(select(User).where(User.email == email)) instead of cur.execute("SELECT * FROM users WHERE email = %s", (email,)). You get type checking, refactor safety, automatic joins, dialect-portability. You lose direct control over the exact SQL emitted — which means you also lose direct knowledge of what is being emitted, unless you go looking.
Most production ORM trouble comes from that loss of knowledge. The fix isn't "stop using the ORM." The fix is turn on SQL echoing in development, read what the ORM generates, and intervene when it's wrong.
engine = create_engine("postgresql+psycopg://app:secret@localhost/myapp", echo=True)
setup added so this can run · defines create_engine
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def create_engine(*_a, **_kw): print('-> create_engine() called') return _AutoMock('create_engine()')
Every query the ORM emits prints to the console. Annoying for a week, life-saving thereafter. Turn it off only in production logs (it's loud) and replace it with sqltap or a per-test query-count assertion.
2. The N+1 Query Problem — The Number-One ORM Killer
The textbook case:
# 1 query for the posts posts = session.scalars(select(Post)).all() # N queries — one per post — for each author for post in posts: print(post.title, "by", post.author.name)
setup added so this can run · defines session, select, Post
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) session = _AutoMock('session') def select(*_a, **_kw): print('-> select() called') return _AutoMock('select()') Post = _AutoMock('Post')
A page of 50 posts fires 51 queries. A page of 500, 501. The latency is dominated by network round trips, each fast on its own, ruinous in bulk. This is the N+1 problem.
The fix is to tell SQLAlchemy to fetch the relationship up front:
from sqlalchemy.orm import joinedload, selectinload # joinedload — SQL JOIN; one query total. Right for one-to-one and many-to-one. posts = session.scalars( select(Post).options(joinedload(Post.author)) ).all() # selectinload — separate IN(...) query for related rows. Right for one-to-many. authors = session.scalars( select(Author).options(selectinload(Author.posts)) ).all()
setup added so this can run · defines session, select, Post, Author
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) session = _AutoMock('session') def select(*_a, **_kw): print('-> select() called') return _AutoMock('select()') Post = _AutoMock('Post') Author = _AutoMock('Author')
Three loaders to choose between:
| Loader | Strategy | Best for |
|---|---|---|
joinedload | SQL JOIN; one query | One-to-one, many-to-one |
selectinload | Separate WHERE id IN (...) query | One-to-many, many-to-many |
subqueryload | Subquery (legacy) | Avoid in new code |
Why not always joinedload? Because joining a parent to a one-to-many child multiplies rows — 10 authors × 100 posts each = 1000 rows shipped over the wire, most of them duplicated parent columns. selectinload issues two queries that return 10 + 1000 rows total. Always two queries; usually faster.
Detect N+1 in tests with sqltap or a counter on Engine.before_cursor_execute:
def test_author_page_query_count(client, engine): count = 0 @event.listens_for(engine, "before_cursor_execute") def _(*args, **kwargs): nonlocal count count += 1 client.get("/authors") assert count <= 3 # if it creeps up, you have a regression
setup added so this can run · defines event
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) event = _AutoMock('event')
3. Fail Loudly on Forgotten Eager Loads
The dangerous part of lazy loading is silence. The relationship works in dev; the query log says "N+1" but nobody's reading it; production gets slow. Make it loud:
class Post(Base): __tablename__ = "posts" id: Mapped[int] = mapped_column(primary_key=True) author_id: Mapped[int] = mapped_column(ForeignKey("authors.id")) author: Mapped["Author"] = relationship(lazy="raise")
setup added so this can run · defines Base, Mapped, mapped_column, relationship, ForeignKey
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) Base = _AutoMock('Base') Mapped = _AutoMock('Mapped') def mapped_column(*_a, **_kw): print('-> mapped_column() called') return _AutoMock('mapped_column()') def relationship(*_a, **_kw): print('-> relationship() called') return _AutoMock('relationship()') def ForeignKey(*_a, **_kw): print('-> ForeignKey() called') return _AutoMock('ForeignKey()')
lazy="raise" makes any access to post.author raise InvalidRequestError unless you explicitly eager-loaded it. The query log becomes irrelevant; the bug becomes a test failure. This is the single most useful relationship default in a serious codebase.
4. Bulk Operations — Don't Loop the Session
session.add() in a loop builds up a unit of work, then flushes individual INSERTs. For one or two objects that's fine. For ten thousand it's catastrophic — one INSERT per row, each with its own SQL parse and round trip.
# BAD — 10,000 INSERTs for u in user_dicts: session.add(User(**u)) session.commit()
setup added so this can run · defines user_dicts, session, User
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) user_dicts = ["alpha", "beta", "gamma"] session = _AutoMock('session') def User(*_a, **_kw): print('-> User() called') return _AutoMock('User()')
The 2.0 idiomatic bulk insert:
from sqlalchemy import insert session.execute(insert(User), user_dicts) # one prepared INSERT, batched bind params session.commit()
setup added so this can run · defines user_dicts, session, User
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) user_dicts = _AutoMock('user_dicts') session = _AutoMock('session') User = _AutoMock('User')
For bulk updates, the same shape:
from sqlalchemy import update session.execute( update(User).where(User.id.in_(ids)).values(active=False) ) session.commit() # One UPDATE statement, regardless of how many rows match.
setup added so this can run · defines session, ids, User
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) session = _AutoMock('session') ids = _AutoMock('ids') User = _AutoMock('User')
For really enormous loads (millions of rows), drop to the database's native bulk loader — Postgres COPY, MySQL LOAD DATA INFILE — via cursor.copy_expert on the underlying DBAPI connection.
5. Pagination — Keyset Over Offset
The default ORM pagination — query.offset(page * size).limit(size) — works fine on the first few pages and rots on the deep ones. To return rows 100,000-100,020, Postgres has to count past the first 100,000. Page 5000 is hundreds of milliseconds slow.
Keyset pagination uses the last seen ID instead:
# First page rows = session.scalars( select(Event).order_by(Event.id).limit(20) ).all() # Next page — pass the last id back as a cursor last_id = rows[-1].id rows = session.scalars( select(Event).where(Event.id > last_id).order_by(Event.id).limit(20) ).all()
setup added so this can run · defines session, Event, select
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) session = _AutoMock('session') Event = _AutoMock('Event') def select(*_a, **_kw): print('-> select() called') return _AutoMock('select()')
Constant-time per page, regardless of how deep the cursor goes. Pair with an index on the ordering column (which you almost certainly have on a primary key). The trade-off: no random-access page numbers — but those were broken on a live-updating table anyway.
For UI tables where users actually need page numbers, cap pagination at the first N pages and direct them to search or filtering instead. Nobody clicks to page 4,827.
6. Connection Pooling at the ORM Level
SQLAlchemy ships its own pool. Configure it explicitly:
engine = create_engine( "postgresql+psycopg://app:secret@db/myapp", pool_size=10, # baseline connections max_overflow=10, # extras when busy; pool_size + max_overflow = hard cap pool_timeout=30, # seconds to wait for a free connection pool_recycle=1800, # recycle connections older than 30 minutes pool_pre_ping=True, # test connection liveness before checkout )
setup added so this can run · defines create_engine
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def create_engine(*_a, **_kw): print('-> create_engine() called') return _AutoMock('create_engine()')
Two of these matter enormously in production:
pool_pre_ping=True— issues a cheapSELECT 1before handing out a connection. Catches connections silently killed by the database, a load balancer idle timeout, or a network blip. Without it, you'll seeOperationalError: server closed the connection unexpectedlyafter every deploy.pool_recycle— forces fresh connections periodically, sidestepping any infrastructure that drops idle connections after some interval (PgBouncer, AWS RDS Proxy, every cloud load balancer).
Total connections per database = pool_size + max_overflow per process × number of processes. Stay under Postgres max_connections (default 100; raise it, or front with PgBouncer; see db-postgres).
7. The Unit-of-Work Pattern
SQLAlchemy's Session is a unit of work — pending writes accumulate, then flush in one transaction. The discipline is to wrap one logical operation in one session lifetime.
with Session(engine) as session, session.begin(): user = User(email="linus@example.com", name="Linus") session.add(user) session.flush() # assigns user.id session.add(AuditLog(user_id=user.id, action="signup")) # commits on clean exit; rolls back on exception
setup added so this can run · defines Session, engine, User, AuditLog
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) def Session(*_a, **_kw): print('-> Session() called') return _AutoMock('Session()') engine = _AutoMock('engine') def User(*_a, **_kw): print('-> User() called') return _AutoMock('User()') def AuditLog(*_a, **_kw): print('-> AuditLog() called') return _AutoMock('AuditLog()')
In a web framework, this is one session per request:
# FastAPI dependency def get_db(): with Session(engine) as session: yield session @app.post("/users") def create_user(payload: UserCreate, session: Session = Depends(get_db)): user = User(**payload.dict()) session.add(user) session.commit() return user
setup added so this can run · defines UserCreate, Session, Depends, User, app, engine
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) UserCreate = _AutoMock('UserCreate') def Session(*_a, **_kw): print('-> Session() called') return _AutoMock('Session()') def Depends(*_a, **_kw): print('-> Depends() called') return _AutoMock('Depends()') def User(*_a, **_kw): print('-> User() called') return _AutoMock('User()') app = _AutoMock('app') engine = _AutoMock('engine')
One session per request, opened on the way in, closed on the way out. Same pattern for Celery tasks (one per task), background workers (one per job), and CLI commands (one per command).
8. Migrations — Append-Only and Zero-Downtime
Two non-negotiable rules.
Migrations are append-only. Once a migration has run in production, you don't edit it. You write a new migration that amends or undoes. Editing applied migrations breaks every environment that's already moved past them.
Zero-downtime schema changes follow expand → backfill → contract. You can't drop a column on a table the running app is reading from. The discipline:
1. Expand — add the new column nullable; deploy code that writes to both old and new.
2. Backfill — populate the new column from the old, in batches, off the request path.
3. Migrate reads — deploy code that reads from the new column.
4. Contract — drop the old column.
Each step is one migration and one deploy. Four deploys to rename a column is the price of never taking the service down.
Alembic is the standard for SQLAlchemy:
alembic revision --autogenerate -m "add user.timezone" alembic upgrade head
Autogenerate is suggestive, not authoritative — review every generated migration before committing. Detection of column-renames and constraint changes is partial.
Django and Flask-Migrate have equivalent workflows; see django-models and flask-database.
9. Optimistic vs Pessimistic Locking
Concurrent writers to the same row need a strategy. Two options.
Optimistic — add a version column. Each update increments it and asserts the previous value:
class Account(Base): id: Mapped[int] = mapped_column(primary_key=True) balance: Mapped[int] version: Mapped[int] = mapped_column(default=0) result = session.execute( update(Account) .where(Account.id == account_id, Account.version == expected_version) .values(balance=new_balance, version=Account.version + 1) ) if result.rowcount == 0: raise ConflictError("someone else updated this account")
setup added so this can run · defines Base, Mapped, mapped_column, session, ConflictError, new_balance, account_id, expected_version, update
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) Base = _AutoMock('Base') Mapped = _AutoMock('Mapped') def mapped_column(*_a, **_kw): print('-> mapped_column() called') return _AutoMock('mapped_column()') session = _AutoMock('session') def ConflictError(*_a, **_kw): print('-> ConflictError() called') return _AutoMock('ConflictError()') new_balance = _AutoMock('new_balance') account_id = _AutoMock('account_id') expected_version = _AutoMock('expected_version') def update(*_a, **_kw): print('-> update() called') return _AutoMock('update()')
No locks; the loser retries. Good for low-contention workloads where conflicts are rare.
Pessimistic — lock the row at read time:
account = session.scalar( select(Account).where(Account.id == account_id).with_for_update() ) account.balance += amount session.commit() # releases the lock
setup added so this can run · defines amount, session, account_id, select, Account
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) amount = 1 session = _AutoMock('session') account_id = _AutoMock('account_id') def select(*_a, **_kw): print('-> select() called') return _AutoMock('select()') Account = _AutoMock('Account')
with_for_update() translates to SQL SELECT ... FOR UPDATE, holding a row-level lock until commit. Right for high-contention workloads where a retry loop would thrash. Watch out for deadlocks when multiple transactions take locks in different orders — always lock rows in a consistent order (e.g., always lock the lower account ID first).
10. Soft Delete — Pay the Cost Deliberately
A common pattern: instead of DELETE FROM users WHERE id = 1, set deleted_at = now() and filter every read. Useful for undeletion, audit, and accidentally-dropping-the-table-recovery. But it leaks complexity:
- Every query needs
WHERE deleted_at IS NULL. Forget it once and you leak deleted data. - Unique constraints break — a re-registered email can't reuse a previously-deleted user's email without a partial unique index.
- Foreign keys reference soft-deleted parents — does the child stay valid?
Implement deliberately or not at all. SQLAlchemy can hide it with a lazy="joined" filter or a global query event, but the schema complexity remains. For audit, often a separate deleted_users table is cleaner.
11. Audit Trails — created_at, updated_at, created_by
A timestamp pair on every table is one of the cheapest debugging investments you'll ever make. Implement with a mixin:
from datetime import datetime from sqlalchemy.orm import Mapped, mapped_column from sqlalchemy import func class TimestampMixin: created_at: Mapped[datetime] = mapped_column(server_default=func.now()) updated_at: Mapped[datetime] = mapped_column( server_default=func.now(), onupdate=func.now() ) class User(Base, TimestampMixin): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) email: Mapped[str]
setup added so this can run · defines Base
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) Base = _AutoMock('Base')
server_default=func.now() lets the database set the timestamp — preferable to Python's clock for consistency. onupdate=func.now() updates the timestamp whenever SQLAlchemy emits an UPDATE for that row.
For who-changed-what, add a created_by / updated_by column populated from the request context, or set up a real audit log table populated by triggers or by application code.
12. Schema Design — Boring Things Done Well
The 80% rules.
- Always a primary key.
BIGSERIAL(cheap, ordered) orUUID(no information leak, distributed-friendly). Don't mix in one schema. - Foreign keys with explicit
ON DELETE—CASCADE,SET NULL, orRESTRICT. Make the choice; don't let it default to "I'll figure it out later".ondelete="RESTRICT"is the safe default — refuses deletes that would orphan children. - Index every foreign key. Postgres does not automatically index FKs. Cascading deletes do full table scans without one.
- Index every column you frequently
WHEREon. Check the query log; checkpg_stat_user_indexesto find unused ones. - Don't over-normalise. A
country_code TEXTdenormalised in the users table is almost always faster and clearer than a join to acountries(id, code)table. Reserve normalisation for data that genuinely changes together.
13. When to Drop to Raw SQL
The ORM is great for CRUD. It's clumsy for:
- Complex reports — window functions, multiple CTEs, recursive queries.
- Database-specific features — Postgres
LATERALjoins,JSONBoperators, full-text search ranking. - Bulk transforms —
INSERT ... SELECTfrom one table to another.
Drop to session.execute(text(...)) without apologising:
from sqlalchemy import text result = session.execute(text(""" SELECT date_trunc('day', created_at) AS day, count(*) AS signups FROM users WHERE created_at > :since GROUP BY 1 ORDER BY 1 DESC """), {"since": since}) for row in result: print(row.day, row.signups)
setup added so this can run · defines session, since
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) session = _AutoMock('session') since = _AutoMock('since')
text() still gets bound parameters — :name style — so injection is still impossible. You give up dialect portability; you keep safety.
14. Testing With the ORM
Two traps.
In-memory SQLite is fast but lies. It doesn't have JSONB. It treats Array columns as strings. CTE semantics differ. LATERAL joins don't exist. Tests pass; production breaks. If your production database is Postgres, your CI database should be Postgres — slow, real, correct.
Transactions per test, not databases per test. Open a transaction at the start of each test, roll it back at the end. Each test sees a clean slate without recreating the schema. The pytest-postgresql and pytest-flask-sqlalchemy plugins automate this.
@pytest.fixture def session(engine): connection = engine.connect() transaction = connection.begin() session = Session(bind=connection) yield session session.close() transaction.rollback() connection.close()
setup added so this can run · defines pytest, Session
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) pytest = _AutoMock('pytest') def Session(*_a, **_kw): print('-> Session() called') return _AutoMock('Session()')
Common Mistakes
1. ORM-everywhere ideology. "We use the ORM" turning into a religious refusal to write SQL for the complex reporting query that the ORM can't express cleanly. Drop to text() and move on.
2. SQL invisible by default. Without echo=True or sqltap, you have no idea what queries are running. Turn it on in dev.
3. Offset pagination on big tables. See Section 5. Use keyset.
4. Per-row session.add() in batch jobs. See Section 4. Use session.execute(insert(...)).
5. Treating the ORM as a black box. Read the SQL it emits. Read it the day you add a new query, not the day it falls over in production.
6. Sharing sessions across threads or tasks. A Session is single-threaded. Use scoped sessions (scoped_session), or — better — pass a session into each task and close it explicitly.
7. Long-lived sessions. A session held for the whole lifetime of a worker accumulates objects in its identity map and pins memory. One session per request/task/job; close it; move on.
8. Committing inside loops over query results. Mutates the cursor's snapshot and produces baffling skipped-row bugs. Iterate first into a list (or use a server-side cursor), then commit.
🎯 Your Turn — Eliminate N+1 from a Slow Function
A teammate wrote the following function. It loads all users, then for each user loads their orders, then for each order loads the product. For 100 users with 10 orders each, that's 1 + 100 + 1000 = 1101 queries. Rewrite it to use three queries total by combining selectinload and joinedload.
Schema (already defined for you):
class User(Base): __tablename__ = "users" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str] orders: Mapped[list["Order"]] = relationship(back_populates="user", lazy="raise") class Order(Base): __tablename__ = "orders" id: Mapped[int] = mapped_column(primary_key=True) user_id: Mapped[int] = mapped_column(ForeignKey("users.id")) product_id: Mapped[int] = mapped_column(ForeignKey("products.id")) user: Mapped["User"] = relationship(back_populates="orders", lazy="raise") product: Mapped["Product"] = relationship(lazy="raise") class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(primary_key=True) name: Mapped[str]
setup added so this can run · defines Base, Mapped, mapped_column, relationship, ForeignKey
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) Base = _AutoMock('Base') Mapped = _AutoMock('Mapped') def mapped_column(*_a, **_kw): print('-> mapped_column() called') return _AutoMock('mapped_column()') def relationship(*_a, **_kw): print('-> relationship() called') return _AutoMock('relationship()') def ForeignKey(*_a, **_kw): print('-> ForeignKey() called') return _AutoMock('ForeignKey()')
Slow version (don't change the schema — change the query):
from sqlalchemy import select def list_orders(session): output = [] users = session.scalars(select(User)).all() # 1 query for user in users: for order in user.orders: # N queries (lazy load) output.append({ "user": user.name, "product": order.product.name, # N×M queries (lazy load per order) }) return output
setup added so this can run · defines User
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) User = _AutoMock('User')
Note that lazy="raise" is already set — the slow version actually raises InvalidRequestError until you fix the eager loading. Good. That's the trap that exists to be sprung.
Rewrite list_orders to fire exactly three SQL queries: one for users, one for their orders, one for the products referenced. Verify with echo=True on the engine while you develop.
Hint 1 — Compose the loaders
You can chain loaders for nested relationships:selectinload(User.orders).joinedload(Order.product). The outer selectinload fetches orders for all loaded users in one IN(...) query; the inner joinedload joins products into that same orders query.
Hint 2 — Why selectinload outer and joinedload inner
Users→orders is one-to-many —selectinload avoids the row-multiplication a JOIN would cause. Order→product is many-to-one — joinedload attaches the product in the same query without multiplying rows. That gives you 1 (users) + 1 (orders + products via JOIN) = 2 queries, plus the original select that triggered the load. Three round trips.
Show full solution
from sqlalchemy import select from sqlalchemy.orm import selectinload, joinedload def list_orders(session): stmt = ( select(User) .options( selectinload(User.orders).joinedload(Order.product) ) ) users = session.scalars(stmt).all() output = [] for user in users: for order in user.orders: output.append({ "user": user.name, "product": order.product.name, }) return output # Demo with echo=True engine — watch the query count # Output should be exactly three SELECTs: # SELECT users.id, users.name FROM users # SELECT orders.id, orders.user_id, orders.product_id, products.id, products.name # FROM orders LEFT OUTER JOIN products ON products.id = orders.product_id # WHERE orders.user_id IN (?, ?, ?, ...) # (the joinedload merges products into the orders fetch — no third query, just 2 total # plus the initial select trigger.)
setup added so this can run · defines User, Order
# Lightweight mock for objects whose attributes/methods aren't critical class _AutoMock: def __init__(self, name='mock'): self._name = name def __getattr__(self, k): return _AutoMock(self._name + '.' + k) def __call__(self, *a, **kw): print('-> ' + self._name + '() called') return _AutoMock(self._name + '()') def __repr__(self): return '<mock ' + self._name + '>' def __str__(self): return '<mock ' + self._name + '>' def __bool__(self): return True def __iter__(self): return iter([]) def __len__(self): return 0 def __getitem__(self, k): return _AutoMock(self._name + '[...]') def __setitem__(self, k, v): pass def __enter__(self): return self def __exit__(self, *a): return False async def __aenter__(self): return self async def __aexit__(self, *a): return False def __add__(self, o): return self def __radd__(self, o): return self def __sub__(self, o): return self def __mul__(self, o): return self def __rmul__(self, o): return self def __truediv__(self, o): return self def __eq__(self, o): return isinstance(o, _AutoMock) def __hash__(self): return hash(self._name) def __lt__(self, o): return True def __le__(self, o): return True def __gt__(self, o): return False def __ge__(self, o): return False def __mro_entries__(self, bases): return (object,) User = _AutoMock('User') Order = _AutoMock('Order')
What the solution gets right:
selectinload(User.orders)— Postgres fetches all orders for all loaded users in oneWHERE user_id IN (...)query. No row multiplication, no per-user query..joinedload(Order.product)— chained onto the inner relationship. Products come back JOINed into the orders query, not as a separate fetch. Many-to-one is exactly whatjoinedloadis built for.lazy="raise"on the relationships — if you forget theoptions(...)you get an immediate exception, not a silent slow page. Bake this into your codebase.- The Python loop is unchanged — the consumer code is identical to the slow version. All the work is in the query options. That's the right place for it.
For an even tighter version on a hot path, drop into a single SQL statement with JOINs and shape the result manually — fewer ORM allocations. But the three-query version is almost always fast enough and is enormously cleaner. Pick the right level of optimisation for the route's actual traffic.
What You Learned
- Turn on
echo=Truein development; you can't fix queries you can't see. - N+1 is the number-one ORM bug. Fix with
joinedload(many-to-one) andselectinload(one-to-many). Chain them for nested relationships. lazy="raise"turns forgotten eager loads into immediate exceptions.- Bulk inserts and updates via
session.execute(insert(...))/update(...)— never per-row in batch jobs. - Keyset pagination over offset on large tables.
- Connection pool with
pool_pre_ping=Trueandpool_recycleto survive infrastructure that kills idle connections. - One session per request/task — open on entry, commit, close.
- Migrations are append-only. Zero-downtime via expand → backfill → contract.
- Optimistic locking (version column) for low contention, pessimistic (
FOR UPDATE) for high contention; lock in consistent order. - Audit columns —
created_at/updated_atvia a mixin, set by the database. - Schema design — primary keys, foreign-key
ON DELETE, index every FK, don't over-normalise. - Drop to raw SQL with
session.execute(text(...))for reports, CTEs, and dialect-specific features. Bind parameters; injection-safe. - Test against real Postgres, not in-memory SQLite, when production is Postgres.
Next: Docker Fundamentals — package your app with its database, so the whole "Docker is the easiest way" line stops being a hand-wave.