ORMs & Database Access Patterns

Object-relational mapping: benefits, pitfalls, and best practices.

The ORM Dilemma

ORMs promise to eliminate SQL and treat databases like in-memory objects. But they come with hidden performance traps, and the expensive ones are invisible in the Python: a single attribute access can turn into a thousand queries. Every measurement in this lesson was taken on PostgreSQL 16 with SQLAlchemy 2.0 so you can see which traps actually cost what. Understanding when to use an ORM, and when to bypass it, is critical for scalable applications.

Real-World Impact:

Stack Overflow wrote and open-sourced Dapper, a micro-ORM that maps query results to objects but leaves you writing the SQL. That is the shape of most mature answers to this problem: keep the mapping, drop the query generation.
The common pattern is not ripping the ORM out. It is keeping it for the 95% of queries where developer speed matters more than milliseconds, and hand-writing the handful of hot queries where it does not.
The traps below are measured rather than quoted, and every script is runnable, so you can reproduce them and see which ones actually apply to your workload.

Try it locally (PostgreSQL 16)

Every script in this lesson runs against the container below. Each one is standalone: it carries its own imports, model definitions and seed data, so you can paste any single block into a file and run it without having run any of the others first.

# Plain PostgreSQL 16 is all this lesson needs.
docker run -d --name pg-demo \
  -e POSTGRES_PASSWORD=demo -e POSTGRES_USER=demo -e POSTGRES_DB=demo \
  -p 5432:5432 postgres:16-alpine

# Clean up when you are done:  docker rm -f pg-demo

# Plus the ORM. Python packages belong in a virtualenv: on Debian/Ubuntu a
# bare "pip install" is refused (PEP 668: externally-managed-environment).
python3 -m venv .venv && source .venv/bin/activate
pip install sqlalchemy "psycopg[binary]"

Popular ORMs Compared

ORMs map database tables to classes and rows to objects. They handle CRUD operations, relationships, and transactions without writing SQL. Let's compare the major players.

ORM	Language	Type	Key Features	Best For
SQLAlchemy	Python	Full ORM + Core	Most flexible, dual APIs (ORM + query builder), mature	Complex queries, data pipelines, enterprise apps
Hibernate	Java	Full ORM	JPA standard, caching layers, lazy collections by default	Large Java enterprise applications
TypeORM	TypeScript/JS	Full ORM	TypeScript-first, decorators, active record pattern	Node.js APIs, TypeScript projects
Prisma	TypeScript/JS	Query Builder	Type-safe, auto-generated client, migration system	Modern Node.js apps, startups, rapid development
Django ORM	Python	Full ORM	Integrated with Django, simple API, admin interface	Django web applications, prototypes

Basic SQLAlchemy Setup

This is the preamble every later script repeats: the two mapped classes, the engine, and a seed() helper that drops, recreates and fills the tables. Starting each example from a known state is what makes the timings below comparable, and the seed data is deliberately deterministic rather than random() so the counts printed later reproduce exactly. Run this one first to confirm the stack is wired up.

Name the driver in the URL. A bare postgresql:// URL selects psycopg2, not the psycopg 3 you just installed, and create_engine() fails with ModuleNotFoundError: No module named 'psycopg2'. Write postgresql+psycopg:// instead.

# setup_check.py - create the schema and confirm the stack works.
from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


n_users, posts_each = 1000, 10
seed(n_users, posts_each)

with engine.connect() as conn:
    users = conn.execute(text("SELECT count(*) FROM users")).scalar()
    posts = conn.execute(text("SELECT count(*) FROM posts")).scalar()

import sqlalchemy, psycopg
print(f"SQLAlchemy {sqlalchemy.__version__}, psycopg {psycopg.__version__}")
print(f"Seeded {users} users and {posts} posts")

Expected Output:

$ python setup_check.py
SQLAlchemy 2.0.51, psycopg 3.3.4
Seeded 1000 users and 10000 posts

SQLAlchemy inspects the mapped classes and generates the CREATE TABLE statements for you. Base.metadata.create_all() is fine for a demo, but it only ever creates what is missing: it will not alter an existing table to match a changed model. Tracking schema changes over time is what Alembic (Lesson 23) is for.

Basic CRUD Operations

With the mapping in place, the four basic operations are ordinary Python. Note that the new user's id is populated by the commit(): the ORM flushes the INSERT, reads back the generated primary key, and writes it onto the object you already hold.

# crud_demo.py - the four basic operations, start to finish.
from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(3, 2)
session = Session()

# CREATE
new_user = User(username="alice", email="alice@example.com")
session.add(new_user)
session.commit()
print(f"created  id={new_user.id} {new_user.username} {new_user.email}")

# READ
alice = session.query(User).filter(User.username == "alice").one()
print(f"read     id={alice.id} {alice.email}")

# UPDATE
alice.email = "newemail@example.com"
session.commit()
print(f"updated  id={alice.id} {alice.email}")

# DELETE
session.delete(alice)
session.commit()
print(f"deleted  remaining users: {session.query(User).count()}")

session.close()

Expected Output:

$ python crud_demo.py
created  id=4 alice alice@example.com
read     id=4 alice@example.com
updated  id=4 newemail@example.com
deleted  remaining users: 3

This is the ORM at its best: readable, type-safe enough for editor autocomplete, and free of string concatenation. The cost is that you can no longer see the queries. Nothing in the code above tells you how many statements were sent, and the rest of this lesson is largely about the cases where that number is much larger than it looks.

The N+1 Query Problem

The most common ORM performance killer. You query N parent records, then each parent triggers 1 additional query for its children, resulting in N+1 queries instead of 2.

The Problem: Lazy Loading by Default

The script below hooks SQLAlchemy's before_cursor_execute event to count statements, so the query count is observed rather than assumed. It loads 1,000 users and asks each one how many posts it has, first with the default strategy and then with each eager-loading option.

# nplus1_demo.py - counts the queries each loading strategy actually issues.
import time

from sqlalchemy import event
from sqlalchemy.orm import joinedload, subqueryload, selectinload

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(1000, 10)

queries = []


@event.listens_for(engine, "before_cursor_execute")
def record(conn, cursor, statement, params, context, executemany):
    queries.append(statement)


def count_posts_per_user(strategy=None, label=""):
    """Load every user, then ask each one how many posts it has."""
    queries.clear()
    session = Session()
    start = time.perf_counter()

    query = session.query(User)
    if strategy is not None:
        query = query.options(strategy)
    for user in query.all():
        len(user.posts)          # <- this attribute access is what costs money

    elapsed = (time.perf_counter() - start) * 1000
    session.close()
    print(f"{label:<26} {len(queries):>5} queries  {elapsed:8.1f} ms")


count_posts_per_user(None, "lazy (default)")
count_posts_per_user(joinedload(User.posts), "joinedload")
count_posts_per_user(subqueryload(User.posts), "subqueryload")
count_posts_per_user(selectinload(User.posts), "selectinload")

Expected Output:

$ python nplus1_demo.py
lazy (default)              1001 queries    1071.5 ms
joinedload                     1 queries      73.4 ms
subqueryload                   2 queries      79.6 ms
selectinload                   3 queries      86.0 ms

Performance Impact: The default strategy issued 1,001 queries and took 1,072 ms. One line of configuration takes it to a single query and 73 ms. Nothing in the Python changed: len(user.posts) looks like an attribute access, and each one was a round trip to PostgreSQL.

What Each Strategy Actually Emits

The three eager-loading options fix the query count in genuinely different ways, and the differences matter when you pick one. Rather than describe the SQL, the script captures it:

# strategies_sql.py - the SQL each eager-loading strategy actually emits.
from sqlalchemy import event
from sqlalchemy.orm import joinedload, subqueryload, selectinload

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(3, 2)

captured = []


@event.listens_for(engine, "before_cursor_execute")
def record(conn, cursor, statement, params, context, executemany):
    captured.append(statement.strip())


for strategy, name in [(joinedload(User.posts), "joinedload"),
                       (subqueryload(User.posts), "subqueryload"),
                       (selectinload(User.posts), "selectinload")]:
    captured.clear()
    session = Session()
    for user in session.query(User).options(strategy).all():
        len(user.posts)
    session.close()

    print(f"=== {name}: {len(captured)} quer{'y' if len(captured) == 1 else 'ies'}")
    for i, sql in enumerate(captured, 1):
        # Collapse the column list; the FROM/WHERE shape is the interesting part.
        head, _, tail = sql.partition("\nFROM ")
        print(f"  [{i}] SELECT ...\n      FROM {tail}")
    print()

Expected Output:

$ python strategies_sql.py
=== joinedload: 1 query
  [1] SELECT ...
      FROM users LEFT OUTER JOIN posts AS posts_1 ON users.id = posts_1.author_id

=== subqueryload: 2 queries
  [1] SELECT ...
      FROM users
  [2] SELECT ...
      FROM (SELECT users.id AS users_id
FROM users) AS anon_1 JOIN posts ON anon_1.users_id = posts.author_id

=== selectinload: 2 queries
  [1] SELECT ...
      FROM users
  [2] SELECT ...
      FROM posts
WHERE posts.author_id IN (%(primary_keys_1)s::INTEGER, %(primary_keys_2)s::INTEGER, %(primary_keys_3)s::INTEGER)

joinedload folds the children into the parent query with a LEFT OUTER JOIN. One round trip, but the parent columns repeat once per child, so a parent with 500 children ships its own row 500 times.
subqueryload re-runs the original query as a subquery and joins the children against it. Two round trips and no duplicated parent columns, but the parent query is executed a second time.
selectinload issues a second SELECT ... WHERE author_id IN (...) using the primary keys it already has. No subquery re-execution and no duplication. It batches the keys, which is why 3 users produced 2 queries here but 1,000 users produced 3: the default batch size is 500.

Which one to reach for: for collections, selectinload is the modern default recommendation in the SQLAlchemy 2.0 documentation, and it is the one to try first. On this dataset it was the slowest of the three (102.1 ms against 69.6 ms), because 10 posts per user is small enough that joinedload's row duplication costs almost nothing. That ranking inverts as the collections grow. Use joinedload for many-to-one and one-to-one, where there is nothing to duplicate.

When Each Loading Strategy Works Best

Summarised, the choice comes down to the cardinality of the relationship and whether you need it on every query. Lazy is the only one that can produce N+1, and it is the default.

Lazy

Load related data only when accessed (default).

When to use:

Relationship rarely accessed
Single parent, not a loop

Danger: Causes N+1 in loops!

Joined

One query, LEFT OUTER JOIN.

When to use:

Many-to-one and one-to-one
One-to-few children
Relationship always needed

Best for: Post → Author

Select IN

Second query, IN on the parent keys.

When to use:

One-to-many collections
Many-to-many
Default choice in 2.0

Best for: User → Posts

Subquery

Second query joins against the original as a subquery.

When to use:

Legacy code already using it
Rarely the best choice today

Note: re-runs the parent query

Detecting N+1 in Development

Counting statements with an event hook is precise but intrusive. During development the faster move is to turn on SQLAlchemy's engine logger and read what it sends. The N+1 signature is unmistakable: the same statement repeated with a different bound parameter.

# logging_demo.py - make the ORM show you every query it runs.
import logging

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(3, 2)

# echo="debug" on create_engine() does the same thing; the logger is easier
# to switch on per-module and to leave off in production.
logging.basicConfig(format="%(levelname)s:%(name)s:%(message)s")
logging.getLogger("sqlalchemy.engine").setLevel(logging.INFO)

session = Session()
for user in session.query(User).all():
    len(user.posts)
session.close()

Expected Output:

$ python logging_demo.py
INFO:sqlalchemy.engine.Engine:BEGIN (implicit)
INFO:sqlalchemy.engine.Engine:SELECT users.id AS users_id, users.username AS users_username, users.email AS users_email
FROM users
INFO:sqlalchemy.engine.Engine:[generated in 0.00008s] {}
INFO:sqlalchemy.engine.Engine:SELECT posts.id AS posts_id, posts.title AS posts_title, posts.content AS posts_content, posts.view_count AS posts_view_count, posts.created_at AS posts_created_at, posts.author_id AS posts_author_id
FROM posts
WHERE %(param_1)s::INTEGER = posts.author_id
INFO:sqlalchemy.engine.Engine:[generated in 0.00007s] {'param_1': 1}
INFO:sqlalchemy.engine.Engine:SELECT posts.id AS posts_id, posts.title AS posts_title, posts.content AS posts_content, posts.view_count AS posts_view_count, posts.created_at AS posts_created_at, posts.author_id AS posts_author_id
FROM posts
WHERE %(param_1)s::INTEGER = posts.author_id
INFO:sqlalchemy.engine.Engine:[cached since 0.001311s ago] {'param_1': 2}
INFO:sqlalchemy.engine.Engine:SELECT posts.id AS posts_id, posts.title AS posts_title, posts.content AS posts_content, posts.view_count AS posts_view_count, posts.created_at AS posts_created_at, posts.author_id AS posts_author_id
FROM posts
WHERE %(param_1)s::INTEGER = posts.author_id
INFO:sqlalchemy.engine.Engine:[cached since 0.002363s ago] {'param_1': 3}
INFO:sqlalchemy.engine.Engine:ROLLBACK

Two details worth noticing in that log. The ORM does not emit SELECT *: it names every mapped column explicitly, which is why adding a column to a model changes the SQL of every query that touches it. And the repeated statement is marked [cached since ...], meaning SQLAlchemy compiled it once and reused it. The compilation was never the expensive part. The three network round trips were.

Lazy vs Eager Loading Strategies

Loading strategies determine when related data is fetched. The right choice depends on access patterns, not just performance.

Configuring Default Loading Strategy

The previous section passed a strategy per query via .options(). You can also set one on the relationship itself, which becomes the default for every query that touches it. Per-query options still win where both apply, so a sensible default here does not lock you in.

# lazy_config.py - set the loading strategy on the relationship itself.
from sqlalchemy import (create_engine, Column, Integer, String, ForeignKey,
                        event, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)

    # lazy="select" is the default: each access issues its own SELECT, which
    # is exactly the N+1 shape when it happens inside a loop.
    posts = relationship("Post", back_populates="author", lazy="select")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"))

    # A post almost always needs its author, and it has exactly one, so
    # folding it into the same query costs one extra JOIN and no extra rows.
    author = relationship("User", back_populates="posts", lazy="joined")


with engine.begin() as conn:
    conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
    conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
Base.metadata.create_all(engine)
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users (id, username, email) VALUES "
                      "(1, 'alice', 'alice@example.com')"))
    conn.execute(text("INSERT INTO posts (id, title, author_id) VALUES "
                      "(1, 'First post', 1), (2, 'Second post', 1)"))

queries = []


@event.listens_for(engine, "before_cursor_execute")
def record(conn, cursor, statement, params, context, executemany):
    queries.append(statement.split()[0])


# Post.author is lazy="joined": fetching posts brings the author along.
session = Session()
queries.clear()
posts = session.query(Post).all()
authors = [p.author.username for p in posts]
print(f"{len(posts)} posts + their authors {authors} -> {len(queries)} query")

# User.posts is lazy="select": the collection is fetched only on access.
queries.clear()
user = session.query(User).first()
print(f"loading the user alone           -> {len(queries)} query")
len(user.posts)
print(f"after touching user.posts        -> {len(queries)} queries")
session.close()

Expected Output:

$ python lazy_config.py
2 posts + their authors ['alice', 'alice'] -> 1 query
loading the user alone           -> 1 query
after touching user.posts        -> 2 queries

The output shows both halves working: fetching the posts brought their authors along in a single query, while loading a user on its own took one query and only issued a second when user.posts was touched. The asymmetry is deliberate and it is the general rule: put lazy="joined" on the many-to-one side, where each row has exactly one related row and the JOIN adds no duplication, and leave the one-to-many side lazy so a query for one post never drags in an author's entire post history.

Lazy Loading Options

Each value below can be used both as a lazy= default on the relationship and as a per-query loader option, with the loader option named slightly differently in a few cases.

Option	Loader option	SQL generated	Use case
`lazy='select'`	`lazyload()`	Separate SELECT per parent, on access	Default. Relationship rarely needed
`lazy='joined'`	`joinedload()`	One query, LEFT OUTER JOIN	Many-to-one, one-to-one
`lazy='selectin'`	`selectinload()`	Second SELECT, `WHERE pk IN (...)`, batched	Collections. Preferred in 2.0
`lazy='subquery'`	`subqueryload()`	Second SELECT joined against the original query	Largely superseded by `selectin`
`lazy='dynamic'`	n/a	None until you execute the returned query	Large collections needing filters
`lazy='raise'`	`raiseload()`	None. Raises `InvalidRequestError`	Force explicit loading, ban N+1

Pattern: Dynamic Loading for Large Collections

Every strategy so far loads the whole collection. When a parent can have tens of thousands of children and you only ever want a slice, lazy="dynamic" hands you a query object instead of a list, so filtering, ordering and pagination happen in the database.

# dynamic_demo.py - lazy="dynamic" for collections too big to load whole.
from datetime import datetime, timedelta

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    # dynamic: User.posts is a query you can refine, not a list.
    posts = relationship("Post", back_populates="author", lazy="dynamic")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    created_at = Column(DateTime, nullable=False)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


with engine.begin() as conn:
    conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
    conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
Base.metadata.create_all(engine)
with engine.begin() as conn:
    conn.execute(text("INSERT INTO users (id, username) VALUES (1, 'alice')"))
    # 50 posts spread deterministically over the last 30 days.
    conn.execute(text("""
        INSERT INTO posts (id, title, created_at, author_id)
        SELECT g, 'Post ' || g, now() - mod(g, 30) * INTERVAL '1 day', 1
        FROM generate_series(1, 50) g"""))

session = Session()
user = session.query(User).first()

print(f"type of user.posts: {type(user.posts).__name__}")

# COUNT(*) in the database: no Post objects are built at all.
print(f"post count:         {user.posts.count()}")

# Filter server-side, then load only the matching rows.
recent = user.posts.filter(
    Post.created_at > datetime.now() - timedelta(days=7)
).order_by(Post.created_at.desc()).all()
print(f"posts last 7 days:  {len(recent)}")

# Pagination without touching the other rows.
page = user.posts.order_by(Post.id).limit(3).all()
print(f"first page ids:     {[p.id for p in page]}")

session.close()

Expected Output:

$ python dynamic_demo.py
type of user.posts: AppenderQuery
post count:         50
posts last 7 days:  15
first page ids:     [1, 2, 3]

The attribute is an AppenderQuery, not a list, so .count() becomes a COUNT(*) and builds no objects at all. The trade-off is that user.posts no longer behaves like a sequence: you cannot iterate it without executing a query, and eager loading it is not possible. Reserve it for genuinely large collections.

When ORMs Hurt Performance

ORMs are not a silver bullet. Certain query patterns are better expressed in raw SQL. Recognizing these cases prevents performance issues. The three below share a root cause: the ORM's unit of work is the object, so any operation whose natural unit is the set pays for objects it does not need.

Problem 1: Bulk Operations

Changing a field on every row is the classic case. The ORM has to load each row, track the change, and write it back. The script below measures that against a single SQL statement, and it restores the data before every timed run for a reason explained just after it.

# bulk_update.py - one UPDATE per object versus one UPDATE, honestly measured.
import statistics
import time

from sqlalchemy import event

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(1000, 10)

updates = []


@event.listens_for(engine, "before_cursor_execute")
def record(conn, cursor, statement, params, context, executemany):
    if statement.strip().upper().startswith("UPDATE"):
        updates.append(len(params) if executemany and params else 1)


def restore():
    """Put the emails back before every timed run. Without this, the second
    run finds every value already at its target, SQLAlchemy's dirty check
    finds nothing to flush, and the benchmark times an empty transaction."""
    with engine.begin() as conn:
        conn.execute(text("UPDATE users SET email = username || '@example.com'"))


def orm_update():
    session = Session()
    for user in session.query(User).all():
        user.email = f"{user.username}@newdomain.com"
    session.commit()
    session.close()


def sql_update():
    session = Session()
    session.execute(text(
        "UPDATE users SET email = CONCAT(username, '@newdomain.com')"))
    session.commit()
    session.close()


def timed(fn, runs=9):
    restore()
    fn()                     # warm-up: SQLAlchemy compiles and caches the statement
    times = []
    for _ in range(runs):
        restore()
        start = time.perf_counter()
        fn()
        times.append((time.perf_counter() - start) * 1000)
    return statistics.median(times)


orm_ms, sql_ms = timed(orm_update), timed(sql_update)
print(f"ORM  (one UPDATE per changed object): {orm_ms:6.1f} ms")
print(f"SQL  (single UPDATE)                : {sql_ms:6.1f} ms")
print(f"ratio {orm_ms / sql_ms:.1f}x")

restore()
updates.clear()
orm_update()
print(f"\nORM sent {len(updates)} UPDATE statement(s), "
      f"{sum(updates)} parameter sets")

print("\nThe trap: re-running WITHOUT restoring the data first")
restore()
for i in range(3):
    updates.clear()
    start = time.perf_counter()
    orm_update()
    print(f"  run {i + 1}: {(time.perf_counter() - start) * 1000:6.1f} ms   "
          f"UPDATE statements={len(updates)}  rows={sum(updates)}")

Expected Output:

$ python bulk_update.py
ORM  (one UPDATE per changed object):   42.3 ms
SQL  (single UPDATE)                :    4.0 ms
ratio 10.7x

ORM sent 1 UPDATE statement(s), 1000 parameter sets

The trap: re-running WITHOUT restoring the data first
  run 1:   40.5 ms   UPDATE statements=1  rows=1000
  run 2:   11.1 ms   UPDATE statements=0  rows=0
  run 3:   25.7 ms   UPDATE statements=0  rows=0

About 10x (it lands between 9x and 11x across runs), and the instrumentation shows the ORM is not as naive as it first looks: it sent a single UPDATE statement with 1,000 parameter sets, one executemany() round trip rather than 1,000. The cost is not network chatter, it is that PostgreSQL still applies 1,000 individual row updates, plus the SELECT and the Python-side change tracking to get there.

Why restore() is not optional. The last block of output is a benchmark that measures nothing. On the second run every email is already userN@newdomain.com, so the assignment changes no value, the dirty check finds nothing to flush, and SQLAlchemy sends zero UPDATE statements. The run still takes 11-26 ms because the SELECT is real, which makes the result look like a plausible warm-cache measurement rather than the no-op it is. Any ORM benchmark that mutates data has to reset it between runs.

Problem 2: Complex Aggregations

Aggregation is where the object-per-row model is least defensible. To add up one integer column, the ORM path builds a full Python object for every row, reads one field from each, and discards the rest.

# aggregation.py - sum a column, in Python versus in the database.
import statistics
import time

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(1000, 10)


def python_sum():
    """Load 10,000 Post objects, then throw away every field but one."""
    session = Session()
    total = sum(post.view_count for post in session.query(Post).all())
    session.close()
    return total


def sql_sum():
    """Let PostgreSQL do the arithmetic and return a single number."""
    session = Session()
    total = session.execute(text("SELECT SUM(view_count) FROM posts")).scalar()
    session.close()
    return total


def timed(fn, runs=25):
    fn()
    times = []
    for _ in range(runs):
        start = time.perf_counter()
        fn()
        times.append((time.perf_counter() - start) * 1000)
    return statistics.median(times)


orm_ms, sql_ms = timed(python_sum), timed(sql_sum)
print(f"ORM  (load 10,000 objects, sum in Python): {orm_ms:6.1f} ms")
print(f"SQL  (SUM(view_count))                   : {sql_ms:6.2f} ms")
print(f"ratio {orm_ms / sql_ms:.0f}x")
print(f"same answer: {python_sum() == sql_sum()}")

Expected Output:

$ python aggregation.py
ORM  (load 10,000 objects, sum in Python):   46.2 ms
SQL  (SUM(view_count))                   :   1.10 ms
ratio 42x
same answer: True

42x, and it is not because PostgreSQL adds integers faster than Python. It is that SUM() returns one row over the wire while the ORM version transfers 10,000 rows and constructs 10,000 objects from them. The gap grows with row count, and it applies to COUNT, AVG, MIN/MAX and every window function equally.

Problem 3: Reporting Queries with Joins

This one is about legibility rather than speed. A grouped report over a join is a first-class SQL construct and an awkward Python expression, and both produce the same plan here. The script runs them side by side and asserts they agree.

# reporting.py - the same report written two ways.
from sqlalchemy import func

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(5, 4)
session = Session()

# 1. ORM expression language: composable, but verbose for a flat report.
orm_rows = session.query(
    User.username,
    func.count(Post.id).label("post_count"),
    func.max(Post.created_at).label("last_post"),
).outerjoin(Post).group_by(User.username).order_by(
    func.count(Post.id).desc(), User.username
).limit(10).all()

print("ORM:")
for username, post_count, last_post in orm_rows:
    days = (datetime.now() - last_post).days
    print(f"  {username:<8} {post_count:>3} posts   most recent {days} days ago")

# 2. Raw SQL: reads exactly like the report it produces.
sql_rows = session.execute(text("""
    SELECT u.username,
           COUNT(p.id)       AS post_count,
           MAX(p.created_at) AS last_post
    FROM users u
    LEFT JOIN posts p ON u.id = p.author_id
    GROUP BY u.username
    ORDER BY post_count DESC, u.username
    LIMIT 10
""")).fetchall()

print("\nRaw SQL:")
for username, post_count, last_post in sql_rows:
    days = (datetime.now() - last_post).days
    print(f"  {username:<8} {post_count:>3} posts   most recent {days} days ago")

print(f"\nidentical results: {orm_rows == sql_rows}")
session.close()

Expected Output:

$ python reporting.py
ORM:
  user1      4 posts   most recent 0 days ago
  user2      4 posts   most recent 1 days ago
  user3      4 posts   most recent 2 days ago
  user4      4 posts   most recent 3 days ago
  user5      4 posts   most recent 4 days ago

Raw SQL:
  user1      4 posts   most recent 0 days ago
  user2      4 posts   most recent 1 days ago
  user3      4 posts   most recent 2 days ago
  user4      4 posts   most recent 3 days ago
  user5      4 posts   most recent 4 days ago

identical results: True

Identical results, and for a report of this size the timing difference is not the point. The SQL version reads in the order the report is described (select these columns, group them this way, sort by that) while the ORM version inverts it and repeats func.count(Post.id) in both the select list and the ordering. When a report grows window functions or CTEs, that gap widens fast. Reach for SQL when the query is the thing you are designing, not an incidental consequence of an object graph.

When to Use Raw SQL

Pulling the three problems together, the split is less about performance than about whether the operation's natural unit is a row or a set.

Prefer Raw SQL For:

Bulk updates/inserts (1000+ rows)
Complex reports with GROUP BY, window functions
Database-specific features (full-text search, JSON ops)
Performance-critical hot paths (sub-10ms requirement)
Analytics queries (doesn't need object mapping)
Migrations and schema changes

Prefer ORM For:

CRUD operations on single records
Business logic requiring object behavior
Validations and constraints in Python
Consistent API across multiple databases
Prototyping and rapid development
Working with relationships (foreign keys)

Query Builders vs Raw SQL

Query builders offer a middle ground: SQL-like syntax with Python composability. No object mapping overhead, but still get parameterized queries and database abstraction.

SQLAlchemy Core (Query Builder)

SQLAlchemy ships the query builder underneath the ORM, and you can use it on its own. A Table describes columns without mapping them to a class, so queries return plain tuples and nothing is instantiated per row.

# core_demo.py - SQLAlchemy Core: tables and queries, no ORM classes.
from sqlalchemy import (create_engine, Table, Column, Integer, String,
                        MetaData, select, text)

engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
metadata = MetaData()

# A Table describes an existing table. Nothing is mapped to a Python class,
# so nothing is instantiated per row: results come back as plain tuples.
users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("username", String(50)),
    Column("email", String(255)),
)


def seed(n_users):
    """Create and fill just the users table this example needs."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})


seed(5)

stmt = select(users).where(users.c.username == "user2")

# The query builder emits an explicit column list and a bound parameter,
# never string interpolation, so there is no SQL injection surface.
print("compiled SQL:")
print(f"  {stmt}".replace("\n", "\n  "))

with engine.connect() as conn:
    print(f"\nresult: {conn.execute(stmt).fetchall()}")

Expected Output:

$ python core_demo.py
compiled SQL:
  SELECT users.id, users.username, users.email
  FROM users
  WHERE users.username = :username_1

result: [(2, 'user2', 'user2@example.com')]

Printing the statement shows what you actually get: an explicit column list and a bound parameter placeholder (:username_1), not the interpolated string the Python resembles. The value travels separately from the SQL text, which is what makes injection structurally impossible rather than merely unlikely.

Comparison: Three Approaches

The same lookup, written at each level, so the difference in what comes back is visible side by side.

# three_ways.py - the same lookup at three levels of abstraction.
from sqlalchemy import Table, MetaData, select

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(5, 2)
session = Session()

# The Core view of the same table, for approach 2.
users_t = Table("users", MetaData(),
                Column("id", Integer, primary_key=True),
                Column("username", String(50)),
                Column("email", String(255)))

# 1. Full ORM: returns mapped objects with identity map and change tracking.
orm_result = session.query(User).filter(User.username == "user2").all()
print(f"1. ORM           [{type(orm_result[0]).__name__} object]  "
      f"-> .email = {orm_result[0].email}")
print(f"                 .posts is a live relationship: "
      f"{len(orm_result[0].posts)} posts")

# 2. Core query builder: returns Row tuples, no objects constructed.
core_result = session.execute(
    select(users_t).where(users_t.c.username == "user2")).fetchall()
print(f"2. Query builder {core_result}")

# 3. Raw SQL. SQLAlchemy 2.0 requires text() around a SQL string; passing a
#    bare str to execute() raises ArgumentError. Parameters stay bound.
raw_result = session.execute(
    text("SELECT * FROM users WHERE username = :username"),
    {"username": "user2"},
).fetchall()
print(f"3. Raw SQL       {raw_result}")

try:
    session.execute("SELECT 1")
except Exception as exc:
    print(f"\nbare string -> {type(exc).__name__}: {str(exc).splitlines()[0]}")

session.close()

Expected Output:

$ python three_ways.py
1. ORM           [User object]  -> .email = user2@example.com
                 .posts is a live relationship: 2 posts
2. Query builder [(2, 'user2', 'user2@example.com')]
3. Raw SQL       [(2, 'user2', 'user2@example.com')]

bare string -> ArgumentError: Textual SQL expression 'SELECT 1' should be explicitly declared as text('SELECT 1')

The ORM returns an object whose .posts is a live relationship it will fetch on demand; the other two return tuples and nothing else. That is the whole trade. Query builders skip the identity map and object construction that make the full ORM expensive on large result sets, landing much closer to raw SQL while keeping composability and parameter safety. How much closer depends entirely on how many rows you materialize, so measure your own hot query rather than trusting a ratio.

Dynamic Query Building

Search endpoints are where query builders earn their keep. The filters are known only at runtime, and the string-concatenation version of this function is the textbook SQL injection vulnerability.

# dynamic_query.py - build a WHERE clause from whichever filters were supplied.
from sqlalchemy import and_

from sqlalchemy import (create_engine, Table, Column, Integer, String,
                        MetaData, select, text)

engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
metadata = MetaData()

# A Table describes an existing table. Nothing is mapped to a Python class,
# so nothing is instantiated per row: results come back as plain tuples.
users = Table(
    "users", metadata,
    Column("id", Integer, primary_key=True),
    Column("username", String(50)),
    Column("email", String(255)),
)


def seed(n_users):
    """Create and fill just the users table this example needs."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})


seed(5)


def search_users(conn, username=None, email=None, limit=10):
    """Compose a query from the filters that were actually provided."""
    stmt = select(users)

    conditions = []
    if username:
        conditions.append(users.c.username.like(f"%{username}%"))
    if email:
        conditions.append(users.c.email == email)

    if conditions:
        stmt = stmt.where(and_(*conditions))

    return conn.execute(stmt.limit(limit)).fetchall()


with engine.connect() as conn:
    print("no filters:")
    for row in search_users(conn, limit=3):
        print(f"  {row}")

    print("\nusername contains 'user1':")
    for row in search_users(conn, username="user1"):
        print(f"  {row}")

    print("\nusername + email:")
    for row in search_users(conn, username="user", email="user3@example.com"):
        print(f"  {row}")

    # The user-supplied value is bound, never interpolated: this is a search
    # for a literal string, not a broken query and not an injection.
    hostile = "'; DROP TABLE users; --"
    print("\nhostile input is treated as a search string:")
    print(f"  matches: {search_users(conn, username=hostile)}")
    print(f"  users table still has {len(search_users(conn, limit=100))} rows")

Expected Output:

$ python dynamic_query.py
no filters:
  (1, 'user1', 'user1@example.com')
  (2, 'user2', 'user2@example.com')
  (3, 'user3', 'user3@example.com')

username contains 'user1':
  (1, 'user1', 'user1@example.com')

username + email:
  (3, 'user3', 'user3@example.com')

hostile input is treated as a search string:
  matches: []
  users table still has 5 rows

A select() is immutable and every method returns a new statement, so conditions can be accumulated in a list and applied at the end without any string handling. The last case is the payoff: a classic injection payload comes back as an ordinary search term that matched nothing, and the table it tried to drop is still there.

ORM Best Practices

Nearly every rule below is a consequence of one idea: the ORM hides how many statements you are sending, so the job is to make that number visible and then keep it small. The habits on the left do that; the ones on the right hide it again.

Do This

Enable query logging in development to catch N+1
Use eager loading (selectinload, joinedload) in loops
Prefer bulk operations for updates/inserts of 100+ rows
Use query builders for read-heavy APIs and reports
Set lazy='raise' during development to force explicit loading
Profile database queries in production (slow query logs)
Use database functions (COUNT, SUM) instead of Python loops
Reset mutated data between benchmark runs, or you measure a no-op

Avoid This

Don't loop over ORM objects to aggregate, use SQL
Don't use ORM for bulk inserts (use executemany or COPY)
Don't ignore N+1 warnings from profiling tools
Don't fetch entire tables into memory (.all() on large tables)
Don't use lazy loading if you always access relationships
Don't build SQL strings manually (SQL injection risk)
Don't add ORM to performance-critical paths without measuring
Don't assume ORM is slower, measure first, optimize if needed

Hybrid Approach: ORM + Raw SQL

In practice the two are not alternatives, they are different tools used in the same session and the same transaction. Writes go through the ORM, where object construction and relationship handling are exactly what you want; the report goes through SQL, where building objects would be pure overhead.

# hybrid.py - ORM for the write, SQL for the report, one transaction.
from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(3, 3)
session = Session()

# Write path: the ORM. Object construction, relationship handling and the
# unit-of-work flush are exactly what it is good at.
author = User(username="alice", email="alice@example.com")
author.posts = [
    Post(title="First post", content="...", view_count=120),
    Post(title="Second post", content="...", view_count=80),
]
session.add(author)
session.commit()
print(f"inserted user {author.id} with {len(author.posts)} posts")

# Read path: SQL. A grouped report needs no objects, so building them is
# pure overhead, and the SQL reads like the report it produces.
rows = session.execute(text("""
    SELECT u.username,
           COUNT(p.id)                    AS post_count,
           COALESCE(SUM(p.view_count), 0) AS total_views
    FROM users u
    LEFT JOIN posts p ON u.id = p.author_id
    GROUP BY u.username
    ORDER BY total_views DESC, u.username
    LIMIT 5
""")).fetchall()

print("\ntop authors by views:")
for username, post_count, total_views in rows:
    print(f"  {username:<8} {post_count:>2} posts  {total_views:>6} views")

session.close()

Expected Output:

$ python hybrid.py
inserted user 4 with 2 posts

top authors by views:
  user3     3 posts     666 views
  user2     3 posts     555 views
  user1     3 posts     444 views
  alice     2 posts     200 views

Assigning a list to author.posts was enough to insert both posts and set their foreign key: the ORM worked out the ordering and the generated id. The report immediately after ran as SQL in the same session, so it sees that write without a second connection or a stale read. Most production applications look like this. There is no need to choose one or the other.

Performance Comparison

One script produces every row of the table below: load 1,000 users and count each one's posts (10,000 posts total), six ways. Timing is the median of nine warm runs, and memory is measured in a separate pass because tracemalloc roughly doubles the runtime of the ORM paths.

# bench_table.py - the numbers in the comparison table below.
import statistics
import time
import tracemalloc

from sqlalchemy import event, func, select
from sqlalchemy.orm import joinedload, subqueryload, selectinload

from datetime import datetime

from sqlalchemy import (create_engine, Column, Integer, String, DateTime,
                        ForeignKey, text)
from sqlalchemy.orm import declarative_base, relationship, sessionmaker

# A bare "postgresql://" URL means psycopg2, which is NOT what
# "pip install psycopg[binary]" gave you. Name the driver explicitly.
engine = create_engine("postgresql+psycopg://demo:demo@localhost:5432/demo")
Session = sessionmaker(bind=engine)
Base = declarative_base()


class User(Base):
    __tablename__ = "users"
    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(255), nullable=False)
    posts = relationship("Post", back_populates="author")


class Post(Base):
    __tablename__ = "posts"
    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String, nullable=False)
    view_count = Column(Integer, nullable=False, default=0)
    created_at = Column(DateTime, nullable=False, default=datetime.now)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")


def seed(n_users, posts_each):
    """Drop, recreate and seed. The data is deterministic, not random, so the
    counts printed below are the same on every run and on your machine too."""
    with engine.begin() as conn:
        conn.execute(text("DROP TABLE IF EXISTS posts CASCADE"))
        conn.execute(text("DROP TABLE IF EXISTS users CASCADE"))
    Base.metadata.create_all(engine)
    with engine.begin() as conn:
        conn.execute(text("""
            INSERT INTO users (id, username, email)
            SELECT g, 'user' || g, 'user' || g || '@example.com'
            FROM generate_series(1, :n) g"""), {"n": n_users})
        conn.execute(text("""
            INSERT INTO posts (id, title, content, view_count, created_at, author_id)
            SELECT g, 'Post ' || g, 'Body of post ' || g,
                   mod(g * 37, 1000),
                   now() - mod(g, 30) * INTERVAL '1 day',
                   mod(g - 1, :n) + 1
            FROM generate_series(1, :p) g"""),
            {"n": n_users, "p": n_users * posts_each})
        conn.execute(text("SELECT setval('users_id_seq', :n)"), {"n": n_users})
        conn.execute(text("SELECT setval('posts_id_seq', :p)"),
                     {"p": n_users * posts_each})
        conn.execute(text("ANALYZE users"))
        conn.execute(text("ANALYZE posts"))


seed(1000, 10)

nq = {"n": 0}


@event.listens_for(engine, "before_cursor_execute")
def count(*args, **kwargs):
    nq["n"] += 1


def orm(strategy):
    def go():
        session = Session()
        query = session.query(User)
        if strategy is not None:
            query = query.options(strategy)
        for user in query.all():
            len(user.posts)
        session.close()
    return go


def core():
    session = Session()
    session.execute(
        select(User.__table__.c.username, func.count(Post.__table__.c.id))
        .select_from(User.__table__.outerjoin(Post.__table__))
        .group_by(User.__table__.c.username)).fetchall()
    session.close()


def raw():
    session = Session()
    session.execute(text("""SELECT u.username, COUNT(p.id)
                            FROM users u LEFT JOIN posts p ON u.id = p.author_id
                            GROUP BY u.username""")).fetchall()
    session.close()


CASES = [
    ("ORM Lazy Loading (N+1)", orm(None)),
    ("ORM Eager (joinedload)", orm(joinedload(User.posts))),
    ("ORM Eager (subqueryload)", orm(subqueryload(User.posts))),
    ("ORM Eager (selectinload)", orm(selectinload(User.posts))),
    ("Query Builder (Core)", core),
    ("Raw SQL", raw),
]

print(f"{'Approach':<28}{'Time':>10}{'Queries':>10}{'Memory':>10}")
results = []
for label, fn in CASES:
    fn()                                # warm up the statement cache
    times = []
    for _ in range(9):
        start = time.perf_counter()
        fn()
        times.append((time.perf_counter() - start) * 1000)

    nq["n"] = 0
    fn()
    queries = nq["n"]

    # Memory is measured in a SEPARATE pass: tracemalloc roughly doubles the
    # runtime of the ORM paths, so timing anything while it is active would
    # overstate every ORM row in this table.
    tracemalloc.start()
    fn()
    peak = tracemalloc.get_traced_memory()[1]
    tracemalloc.stop()

    median = statistics.median(times)
    results.append((label, median))
    print(f"{label:<28}{median:>8.1f}ms{queries:>10}{peak / 1024 / 1024:>8.2f}MB")

base = results[0][1]
print("\nspeedup vs lazy: " + "  ".join(
    f"{label.split('(')[-1].rstrip(')')} {base / ms:.1f}x"
    for label, ms in results[1:]))

Expected Output:

$ python bench_table.py
Approach                          Time   Queries    Memory
ORM Lazy Loading (N+1)        1084.9ms      1001   13.48MB
ORM Eager (joinedload)          69.6ms         1   16.86MB
ORM Eager (subqueryload)        82.8ms         2   15.99MB
ORM Eager (selectinload)       102.1ms         3   16.00MB
Query Builder (Core)             3.6ms         1    0.13MB
Raw SQL                          3.4ms         1    0.13MB

speedup vs lazy: joinedload 15.6x  subqueryload 13.1x  selectinload 10.6x  Core 302.9x  Raw SQL 323.7x

Approach	Time	Queries	Peak memory	Use Case
ORM Lazy Loading (N+1)	1084.9 ms	1,001	13.48 MB	Avoid in production
ORM Eager (joinedload)	69.6 ms	1	16.86 MB	Best here, but duplicates parent rows
ORM Eager (subqueryload)	82.8 ms	2	15.99 MB	Superseded by selectinload
ORM Eager (selectinload)	102.1 ms	3	16.00 MB	Safest default for collections
Query Builder (Core)	3.6 ms	1	0.13 MB	Fast reads, no objects
Raw SQL	3.4 ms	1	0.13 MB	Maximum performance

Key Takeaway: Eager loading fixed the query count: 1,001 down to 1-3, and 1085 ms down to 70-102 ms, roughly 10-16x from a one-line change. But the last two rows did something different. They stopped returning rows at all: one grouped query, no ORM objects, 3.4 ms and 0.13 MB against 13.48 MB. That is ~320x, and it is the larger lesson. Eager loading fixes how many times you ask; pushing work into SQL fixes how much comes back, and the second is worth far more. Note also that eager loading raised peak memory (13.48 → 16.86 MB): it trades memory for round trips, which is the right trade here but not an unconditional win.

About these numbers. Measured on a 12th Gen Intel Core i7-12700H with PostgreSQL 16.14 in Docker on localhost, Python 3.12, SQLAlchemy 2.0.51 and psycopg 3.3.4. Absolute timings are hardware, cache and configuration dependent, and a real network hop between the application and the database would inflate the N+1 row far more than the others, since it is the only one paying 1,001 round trips. What reproduces is the shape and the ordering, not the exact milliseconds. Take a median of several runs on your own data before treating any specific multiple as a target.

Decision Framework

The measurements above do not say "use the ORM" or "use SQL", they say the choice is per query. Walk these five questions for a given access path and the answer usually falls out. Note that they are ordered by how strongly they decide the matter, so stop at the first one that gives a clear answer.

Which Approach Should I Use?

1. Is it a single-record CRUD operation?

YES → Use Full ORM (User, Post objects with relationships)
NO → Continue to question 2

2. Does it involve 100+ rows (bulk operation)?

YES → Use Raw SQL with bulk operations (UPDATE, INSERT ... SELECT)
NO → Continue to question 3

3. Is it a complex analytical query (GROUP BY, window functions)?

YES → Use Raw SQL or Query Builder
NO → Continue to question 4

4. Do you need object behavior (methods, validations)?

YES → Use Full ORM with eager loading if accessing relationships
NO → Use Query Builder (faster, returns tuples/dicts)

5. Is response time critical (<10ms requirement)?

YES → Use Raw SQL + caching
NO → ORM or Query Builder is fine

Pro Tip: Start with ORM for prototyping. Profile in production. Optimize hot paths with query builders or raw SQL only when measurements show it's needed. Premature optimization wastes time, let data guide your decisions.

Key Takeaways

Name the driver in the URL - a bare postgresql:// means psycopg2. With psycopg 3 installed you need postgresql+psycopg:// or create_engine() raises ModuleNotFoundError.
N+1 is invisible in the Python - len(user.posts) inside a loop cost 1,001 queries and 1,072 ms against 73 ms for the same loop with eager loading. Only the query log or a statement counter shows it.
Prefer selectinload for collections - it issues a batched WHERE pk IN (...) with no row duplication and no re-executed subquery. Use joinedload for many-to-one and one-to-one, where there is nothing to duplicate.
Eager loading fixes the query count, not the payload - it took 1085 ms to ~70 ms, and it raised peak memory. Replacing the whole access with one grouped SQL query took it to 3.4 ms and 0.13 MB, roughly 320x.
Aggregate in the database - summing a column over 10,000 rows took 46.2 ms via Python objects and 1.10 ms via SUM(), about 42x, because one row crosses the wire instead of 10,000.
Bulk writes are ~10x, not 100x - SQLAlchemy already batches per-object updates into a single executemany(). The cost is PostgreSQL applying 1,000 row updates plus the SELECT and change tracking to get there.
Reset mutated data between benchmark runs - re-running the bulk update without restoring the emails emitted zero UPDATE statements and still reported a plausible-looking 11 ms. A fast number is not automatically a real one.
Query builders are the middle ground - SQLAlchemy Core gives you bound parameters and composable, injection-proof dynamic queries without the identity map or per-row object construction.
The choice is per query, not per project - ORM for writes and single-record CRUD, SQL for reports and bulk work, in the same session and the same transaction.

Back to "Data Modeling Patterns"

Databases AdvancedLesson 25

Next: "Replication & High Availability"