ORM Patterns That Don't Blow Up in Production
1 · The lesson
readAn ORM is easy to misuse. Used well, it saves weeks of work on a project and keeps the model code readable as the schema grows. Used badly, it produces very slow SQL: N+1 queries, full table scans on every page, updates without transactions, and migrations that lock the database for fifteen minutes in the middle of a deploy.
This lesson covers the patterns and anti-patterns that decide whether your ORM-based service stays fast at 10× your current load, or fails on the day traffic suddenly grows. Examples use SQLAlchemy 2.0, a widely used Python ORM. The same patterns work in the Django ORM and Tortoise ORM with small syntax changes.
Examples assume a running local Postgres. Docker is the easiest way to get one. See devops-docker. Pip-install the driver as shown. Expected output is in comments.
1. The ORM Trade-Off
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, safer refactoring, automatic joins, and code that works on more than one database. You lose direct control over the exact SQL that runs. That also means you don't know what SQL runs, unless you check.
Most production ORM trouble comes from that loss of knowledge. The fix is not to stop using the ORM. The fix is to turn on SQL echoing in development, read the SQL the ORM generates, and change your code when the SQL is 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 runs is printed to the console. The output is annoying at first, but it shows you problems you would otherwise miss. Turn it off only in production (it produces a lot of log output), and use something else there, such as sqltap or a per-test query-count assertion.
2. The N+1 Query Problem: The Most Common ORM Bug
The standard example:
# 1 query for the posts posts = session.scalars(select(Post)).all() # N more queries (one per post) to load 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. Most of the time is spent on network round trips. Each one is fast, but hundreds of them make the page slow. 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: a SQL JOIN, so one query in total. Best for one-to-one and many-to-one. posts = session.scalars( select(Post).options(joinedload(Post.author)) ).all() # selectinload: a second IN(...) query for the related rows. Best 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 sent over the network, and most of each row is a repeated copy of the author's columns. selectinload sends two queries that return 10 + 1000 rows in total, without the repeated columns. It is always two queries, and it is usually faster.
You can 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 the count goes up, a code change added queries
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 danger of lazy loading is that it gives no warning. The relationship works in development. The query log shows the N+1 queries, but nobody reads it. Then production gets slow. Make the mistake raise an error instead:
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. You no longer depend on someone reading the query log, because the bug becomes a test failure. In a large codebase, this is one of the most useful defaults for relationships.
4. Bulk Operations: Don't Call the Session in a Loop
session.add() in a loop builds up a unit of work, and the flush then writes the rows. For one or two objects that's fine. For ten thousand, it is slow: the session creates and tracks ten thousand Python objects, and on some databases it also sends one INSERT per row. (Since SQLAlchemy 2.0, the flush can group these INSERTs into batches on databases that support RETURNING, but whether it does depends on the database and the table.)
# SLOW: 10,000 objects, each tracked by the session 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 standard bulk insert in SQLAlchemy 2.0:
from sqlalchemy import insert session.execute(insert(User), user_dicts) # one prepared INSERT, with the values sent in batches 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')
Bulk updates work the same way:
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 very large loads (millions of rows), use the database's own bulk loader (Postgres COPY, MySQL LOAD DATA INFILE) through cursor.copy() on the underlying psycopg 3 connection (copy_expert in the older psycopg2).
5. Pagination: Keyset Instead of Offset
The usual ORM pagination, query.offset(page * size).limit(size), works fine on the first few pages and gets slower on later ones. To return rows 100,000-100,020, Postgres has to read and skip the first 100,000. Page 5000 can take hundreds of milliseconds.
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()')
Every page takes about the same time, however deep you go. Use it with an index on the ordering column (a primary key already has one). The trade-off is that you can't jump straight to page N. On a table that changes all the time, page numbers were unreliable anyway, because rows move between pages.
For UI tables where users really need page numbers, allow only the first N pages and send users to search or filters for the rest. Nobody clicks to page 4,827.
6. Connection Pooling at the ORM Level
SQLAlchemy has its own connection pool. Configure it explicitly:
engine = create_engine( "postgresql+psycopg://app:secret@db/myapp", pool_size=10, # connections kept open normally max_overflow=10, # extra ones when busy, up to pool_size + max_overflow in total pool_timeout=30, # seconds to wait for a free connection pool_recycle=1800, # recycle connections older than 30 minutes pool_pre_ping=True, # check that the connection still works before using it )
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 settings are very important in production:
pool_pre_ping=True: runs a cheapSELECT 1before giving out a connection. This finds connections that were closed without warning by the database, by a load balancer's idle timeout, or by a short network failure. Without it, you'll seeOperationalError: server closed the connection unexpectedlyafter every deploy.pool_recycle: replaces connections after a set time, so they are not dropped by infrastructure that closes idle connections after a while (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 it with PgBouncer (see db-postgres).
7. The Unit-of-Work Pattern
SQLAlchemy's Session is a unit of work: pending writes are collected, then flushed in one transaction. The rule is to use one session for one logical operation.
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 if the block ends normally, rolls back on an 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.model_dump()) # Pydantic 2 name (.dict() is deprecated) session.add(user) session.commit() session.refresh(user) # commit expires the object; load it again return {"id": user.id, "email": user.email, "name": user.name}
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 when the request starts and closed when it ends. 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 rules you must always follow.
Migrations are append-only. Once a migration has run in production, you don't edit it. You write a new migration that changes or undoes it. Editing a migration that has already run breaks every environment where it ran.
Zero-downtime schema changes follow expand → backfill → contract. You can't drop a column while the running app still reads it. The steps:
1. Expand: add the new column as nullable, then deploy code that writes to both old and new.
2. Backfill: fill the new column from the old one, in batches, in a background job rather than during requests.
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 cost of never taking the service down.
Alembic is the standard for SQLAlchemy:
alembic revision --autogenerate -m "add user.timezone" alembic upgrade head
Autogenerate gives you a draft, not a final answer, so review every generated migration before committing it. It does not detect every column rename or constraint change.
Django and Flask-Migrate have equivalent workflows. See django-models and flask-database.
9. Optimistic vs Pessimistic Locking
When several writers can change the same row at the same time, you need a strategy. There are two options.
Optimistic: add a version column. Each update increases it by one, and only succeeds if the version is still the value it read:
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()')
There are no locks. If two writers conflict, the second one retries. Good for low-contention workloads where conflicts are rare.
Pessimistic: lock the row when you read it:
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. It suits workloads with many conflicts, where a retry loop would retry again and again. Watch out for deadlocks when several 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: Know What It Costs
A common pattern: instead of DELETE FROM users WHERE id = 1, set deleted_at = now() and filter every read. It is useful for undoing deletes, for audits, and for recovering data deleted by mistake. But it adds complexity everywhere:
- Every query needs
WHERE deleted_at IS NULL. Forget it once, and deleted data appears in your results. - Unique constraints break: a re-registered email can't reuse a previously-deleted user's email without a partial unique index.
- Foreign keys can point to soft-deleted parents. Is the child still valid?
Decide on it on purpose, or don't use it. SQLAlchemy can add the filter for you, with with_loader_criteria() inside a do_orm_execute session 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
Two timestamp columns on every table cost very little and make debugging much easier. 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. This is better than Python's clock, because every row gets its time from the same clock. 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: The Basics Done Well
The rules that cover most cases:
- Always a primary key.
BIGSERIAL(cheap, ordered) orUUID(doesn't reveal how many rows exist, easy to generate on many machines). Don't mix the two in one schema. - Foreign keys with explicit
ON DELETE:CASCADE,SET NULLorRESTRICT. Make the choice now, not later.ondelete="RESTRICT"is the safe default: it refuses deletes that would leave children without a parent. - Index every foreign key. Postgres does not index FKs automatically. Without an index, cascading deletes do full table scans.
- Index every column you frequently
WHEREon. Check the query log, and 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. Normalise only data that really 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.
In these cases, use session.execute(text(...)):
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 uses bound parameters (:name style), so SQL injection is still not possible. You lose portability between databases, but you keep the safety.
14. Testing With the ORM
Two traps.
In-memory SQLite is fast but behaves differently. Postgres features such as JSONB operators (for example @>), ARRAY columns and LATERAL joins don't exist in SQLite, so code that uses them can't be tested there. Smaller differences let wrong code pass: SQLite accepts the string 'abc' in an INTEGER column, for example. Tests pass, and then production breaks. If your production database is Postgres, your CI database should be Postgres too. It is slower, but the results are 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 starts with clean data, 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. Refusing to write any SQL. "We use the ORM" becomes a strict rule, even for a complex reporting query that the ORM can't express clearly. Use text() for that query.
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 on the day you add a new query, not on the day it fails 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 keeps collecting objects in its identity map, and that memory is never freed. Use one session per request, task or job, and close it when done.
8. Committing inside loops over query results. Each commit is a separate transaction, and by default it expires every object the session has loaded, so each later object is loaded again with an extra SELECT. If the job fails halfway, half the rows are already committed. Change the rows in the loop, then commit once after it (or once per batch).
🎯 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 two 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, so the slow version raises InvalidRequestError until you fix the eager loading. This is the setting from Section 3 doing its job.
Rewrite list_orders to fire exactly two SQL queries: one for users, and one for their orders with the products joined in. 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, and the inner joinedload joins products into that same orders query.
Hint 2: Why selectinload outer and joinedload inner
Users→orders is one-to-many, soselectinload avoids the row-multiplication a JOIN would cause. Order→product is many-to-one, so joinedload attaches the product in the same query without multiplying rows. That gives you 1 (users) + 1 (orders + products via JOIN) = 2 queries in total.
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 an echo=True engine: watch the query count # Output should be exactly two 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, so there is no third query.)
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): SQLAlchemy 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.joinedloadis designed for many-to-one relationships like this.lazy="raise"on the relationships: if you forget theoptions(...)you get an immediate exception, not a slow page with no warning. Use it across your codebase.- The Python loop is unchanged: the code that uses the data is identical to the slow version. All the changes are in the query options, which is where they belong.
For a route that runs very often, you can go further: write a single SQL statement with JOINs and build the result yourself, so the ORM creates fewer objects. But the two-query version is almost always fast enough, and it is much easier to read. Optimise only as much as the route's real traffic needs.
What You Learned
- Turn on
echo=Truein development, because 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 with
session.execute(insert(...))/update(...), never one row at a time 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 or task: open it at the start, 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_atfrom 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. Bound parameters keep it safe from injection. - Test against real Postgres, not in-memory SQLite, when production is Postgres.
Next: Docker Fundamentals: package your app with its database, and learn how the "Docker is the easiest way" setup in these lessons works.
Found this useful? Share it with a friend who’s learning Python.