Bulk Operations and Performance

ORM per-row INSERT/UPDATE is slow at scale. For 10k+ rows, switch to Core constructs that batch into single INSERT/UPDATE statements with parameter arrays.

The slow way: ORM per-row

with Session(engine) as session:
    for row in big_list:
        session.add(User(name=row["name"], email=row["email"]))
    session.commit()

What happens:

• One INSERT per row (autoflush may batch a few, but still per-row roughly).
• Identity map populated with N instances → memory growth.
• Cascades, events, defaults all evaluated per row.

10k rows: ~30 seconds, ~500 MB Python memory. Don’t do this.

The fast way: Core insert with executemany

from sqlalchemy import insert

session.execute(
    insert(User),
    [{"name": "alice", "email": "a@b.com"}, {"name": "bob", "email": "b@c.com"}, ...],
)
session.commit()

This uses the DBAPI’s executemany(), which batches all the rows into one round trip. For psycopg/asyncpg on Postgres, this further compiles to a single multi-row INSERT under the hood.

10k rows: ~1 second, minimal memory. 30x faster.

Trade-offs:

• No Python-side defaults run (unless you supply them in the dicts).
• No relationships maintained (no cascades).
• No identity map (you don’t get instances back).
• No event hooks (before_insert, etc.).

If you need server-defaults (like created_at = NOW()), Postgres / SQLite handle them server-side. Python default= callables don’t run.

RETURNING for bulk inserts

For getting the auto-generated PKs:

stmt = insert(User).returning(User.id)
result = session.execute(
    stmt,
    [{"name": "alice"}, {"name": "bob"}],
)
ids = [row.id for row in result]

Available on Postgres and SQLite. For older MySQL, no RETURNING; you’d query last_insert_id.

Bulk update

session.execute(
    update(User),
    [
        {"id": 1, "email": "a@new.com"},
        {"id": 2, "email": "b@new.com"},
    ],
)
session.commit()

Equivalent to one UPDATE per row but batched. Requires the WHERE clause’s column (here id) to be present in each dict.

For “update all matching rows to the same value,” use the simpler Core form:

session.execute(
    update(User)
    .where(User.last_login < cutoff)
    .values(is_active=False)
)

One UPDATE statement affecting all matching rows. Fastest possible.

Bulk delete

session.execute(delete(User).where(User.deleted_at < cutoff))

One DELETE. The ORM’s session.delete(user) is for “this loaded object” — per-row.

ORM bulk_insert_mappings (legacy 1.x, still works)

session.bulk_insert_mappings(User, [{"name": "alice"}, {"name": "bob"}])
session.commit()

Mostly superseded by session.execute(insert(User), [...]) in 2.0. Slight differences: bulk_insert_mappings skips more ORM machinery (faster but more restrictive). Stick with 2.0 style for new code.

ORM session.scalars(insert(...).returning(User)) — bulk insert into instances

stmt = insert(User).returning(User).values([
    {"name": "alice"}, {"name": "bob"}
])
result = session.scalars(stmt)
users = result.all()             # list of User instances

Postgres only (needs RETURNING). Bridges “bulk performance” and “I want instances back.” Doesn’t run Python-side defaults; does run column types and the basic ORM mapping.

Streaming large reads — yield_per

For “process 10M rows without loading them all”:

stmt = select(User).execution_options(yield_per=1000)
for user in session.scalars(stmt):
    process(user)

SQLAlchemy fetches in batches of 1000 rows, holding the cursor open. Memory stays roughly constant.

Server-side cursor (Postgres):

stmt = select(User).execution_options(stream_results=True, yield_per=1000)

Postgres uses a named server-side cursor instead of pulling everything into the client at once. Required for huge result sets where even one batch wouldn’t fit.

Watch out: server-side cursors hold a DB transaction open the whole time. Don’t run an interactive process between rows.

Avoid len(session.scalars(...).all())

# WRONG — loads everything into memory
count = len(session.scalars(select(User)).all())

# RIGHT — single COUNT(*) query
count = session.scalar(select(func.count(User.id)))

The “I just want a count” mistake. Same for “is there at least one”:

# WRONG
exists = session.scalars(select(User).where(...)).first() is not None

# RIGHT
exists = session.scalar(select(User.id).where(...).limit(1)) is not None

# or
from sqlalchemy import exists as sa_exists
exists = session.scalar(select(sa_exists(select(User).where(...))))

Avoid .options() after .execute()

# WRONG — options ignored
stmt = select(User)
users = session.scalars(stmt).all()
users_with_posts = stmt.options(selectinload(User.posts))   # creates new stmt, but already executed

# RIGHT
stmt = select(User).options(selectinload(User.posts))
users = session.scalars(stmt).all()

.options() must be on the statement before execution.

Avoid expensive column_property subqueries

class User(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    post_count: Mapped[int] = column_property(
        select(func.count(Post.id))
        .where(Post.author_id == id)
        .scalar_subquery()
    )

Every SELECT on User now includes a correlated subquery for post_count. If you query 1000 users, 1000 subqueries run. Sometimes worth it for convenience; usually better as a per-query annotation:

stmt = (
    select(User, func.count(Post.id).label("post_count"))
    .outerjoin(Post)
    .group_by(User.id)
)

ORM event listeners can secretly slow things down

@event.listens_for(User, "before_insert")
def before_insert(mapper, connection, target):
    target.something = compute_something(target)   # runs per row

Fine for occasional inserts; per-row evaluation during a 100k-row bulk import is a problem. Use event.listen with care; bulk operations bypass ORM events anyway.

Indexes (the obvious thing people forget)

Most SQLAlchemy “perf problems” are missing or wrong indexes at the DB layer, not the ORM.

class User(Base):
    email: Mapped[str] = mapped_column(unique=True, index=True)
    org_id: Mapped[int] = mapped_column(index=True)

    __table_args__ = (
        Index("ix_user_org_status", "org_id", "status"),   # composite
        Index("ix_user_email_lower", func.lower("email")), # functional
    )

Run EXPLAIN ANALYZE on slow queries. See ../10_explain_analyze.md and ../11_index_types.md.

Connection pool tuning

A small pool is the bottleneck under high concurrency. A huge pool overwhelms the DB. See 10_connection_pooling.md.

echo=False in production

engine = create_engine("...", echo=True)

echo=True logs every SQL statement at INFO. Helpful in dev; catastrophic in production (gigabytes of logs, ~10% performance hit). Use Python logging at WARNING level for production:

import logging
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)

When to drop down to raw SQL

If a query is hard to express in select() or you’ve profiled and the ORM overhead matters:

result = session.execute(
    text("SELECT u.id, count(p.id) FROM users u LEFT JOIN posts p ON ... WHERE ...")
)

text() is parameterized:

session.execute(text("SELECT * FROM users WHERE id = :id"), {"id": 42})

Always use bound parameters; never f-string raw SQL with user input (SQL injection).

For getting models back from raw SQL: select(User).from_statement(text(...)) lets ORM-map the result.

Profiling

from sqlalchemy import event
import time

@event.listens_for(engine, "before_cursor_execute")
def before(conn, cursor, statement, params, context, executemany):
    context._query_start = time.perf_counter()

@event.listens_for(engine, "after_cursor_execute")
def after(conn, cursor, statement, params, context, executemany):
    elapsed = time.perf_counter() - context._query_start
    if elapsed > 0.1:        # 100ms
        logger.warning(f"slow query ({elapsed:.2f}s): {statement[:200]}")

Surface slow queries in logs. For real profiling, pg_stat_statements + Datadog APM / Sentry performance is better.

Common pitfalls

• session.add() in a tight loop for 100k rows — see “the slow way” above.
• N+1 in serialization — loading user, then accessing user.profile.bio for each. Eager load.
• Loading huge result sets into memory — use yield_per and stream.
• column_property with subquery running on every base SELECT — switch to per-query annotation.
• expire_on_commit=True + reading attributes after commit in a loop — SELECT per object.

Common interview confusions

• “ORM is always slower than Core.” — for single-row CRUD, the difference is microseconds. For bulk ops, ORM is much slower.
• “session.bulk_save_objects() is the modern way.” — that’s 1.x. 2.0 way is session.execute(insert(Model), list_of_dicts).
• “Setting indexes in SQLAlchemy creates them at the DB.” — only if you run Base.metadata.create_all(engine). In production, indexes come from Alembic migrations.

Interview angle

• “How do you bulk insert 100k rows efficiently?” — session.execute(insert(Model), list_of_dicts). Uses DBAPI executemany; one round trip; bypasses ORM per-row overhead. 30x+ faster than session.add() loop.
• “What’s the trade-off when using Core bulk insert?” — no ORM event hooks, no relationship cascades, no Python-side defaults (server defaults still work), no instances returned (unless you add .returning()).
• “How do you stream a huge SELECT result without loading it all?” — select(...).execution_options(yield_per=1000) for batched fetching. For Postgres, add stream_results=True for server-side cursor.
• “How do you count rows efficiently?” — session.scalar(select(func.count(Model.id))). Never len(session.scalars(...).all()) — that materializes everything.
• “How would you find slow queries in production?” — before/after_cursor_execute event listeners with a threshold; or pg_stat_statements; or APM tools (Datadog, Sentry, New Relic). Log slow queries with the statement and bind parameters.
• “When would you drop to text() raw SQL?” — complex queries that don’t map cleanly (window functions over multiple tables, recursive CTEs), profiling shows ORM construction overhead matters, or DB-specific syntax SQLAlchemy doesn’t expose. Always use bound parameters.