`bulk_insert_mappings` and its limitations

bulk_insert_mappings (in SQLAlchemy 1.x), or equivalently session.execute(insert(Model), [...]) (in 2.0), is the sweet spot between speed and ORM compatibility. It's 3-5x faster than a loop, keeps compatibility with most of your SQLAlchemy code, but skips ORM events that can be critical.

In this capsule you'll learn when it's the right tool, what exactly it skips (events, Python defaults, validators), and how to verify you're not breaking existing functionality when you switch from add_all to bulk insert.

What it does exactly

In SQLAlchemy 2.0, the canonical pattern is:

from sqlalchemy import insert

await session.execute(
    insert(Task),
    [
        {"title": "T1", "status": "pending"},
        {"title": "T2", "status": "completed"},
        # ...
    ]
)
await session.commit()

SQLAlchemy 2.0+ implements this with the "insertmanyvalues" optimization: it groups rows into fewer SQL statements, reducing round-trips dramatically.

In 1.x the equivalent was session.bulk_insert_mappings(Task, mappings). The API changed but the concept is identical.

What it skips

1. ORM events (@event.listens_for)

from sqlalchemy import event


@event.listens_for(Task, 'before_insert')
def set_default_priority(mapper, connection, target):
    if target.priority is None:
        target.priority = calculate_default_priority()

With add() + commit(): the listener fires, priority gets set. With insert(Task), [...]: the listener does not fire. Tasks are inserted with priority = NULL (or the SQL default if one exists).

2. Validators (@validates)

from sqlalchemy.orm import validates


class Task(Base):
    @validates('title')
    def validate_title(self, key, value):
        if len(value) > 200:
            raise ValueError("Title too long")
        return value.strip()

With add(): validate_title runs, validates and normalizes. With bulk insert: the validator does not run. If the data is invalid, it goes to the DB as-is (probably failing on DB-level constraints, or passing if there's no constraint).

3. Python defaults (default=lambda: ...)

class Task(Base):
    created_at: Mapped[datetime] = mapped_column(
        default=lambda: datetime.now(timezone.utc)
    )
    # vs
    server_created_at: Mapped[datetime] = mapped_column(
        server_default=func.now()
    )

With add(): default=lambda runs in Python, sets the timestamp. With bulk insert: the Python default does not run. If the column isn't sent explicitly and has no server_default, NULL is inserted (error if the column is NOT NULL).

server_default (SQL) does work with bulk because the DB is the one applying the default.

4. Identity map / unique objects

task1 = Task(title="T1")
task2 = task1  # same object
session.add(task1)
session.add(task2)
session.flush()  # Only 1 INSERT — the identity map detects it's the same object

Bulk insert has no identity map. If you pass the same dict twice:

data = [{"title": "T1"}, {"title": "T1"}]  # Identical dicts
await session.execute(insert(Task), data)
# 2 INSERTs executed

5. RETURNING IDs (with caveats)

add() + flush(): SQLAlchemy runs INSERT...RETURNING and populates task.id automatically.

Bulk insert: it depends on the version and dialect. In SQLAlchemy 2.0 with PostgreSQL, insert(Task).returning(Task.id) does return IDs:

result = await session.execute(
    insert(Task).returning(Task.id),
    [{"title": "T1"}, {"title": "T2"}]
)
ids = [row.id for row in result]

But performance is worse because PostgreSQL returns rows per batch. For bulk loads of 100k+ rows, avoid RETURNING.

When you SHOULD use bulk_insert_mappings

Case 1: the model has no critical events or validators.

class SimpleEvent(Base):
    id: Mapped[int] = mapped_column(primary_key=True)
    event_type: Mapped[str] = mapped_column(String(50))
    payload: Mapped[dict] = mapped_column(JSONB)
    created_at: Mapped[datetime] = mapped_column(
        server_default=func.now()  # ← server-side default
    )

No events, no validators, defaults are SQL. Bulk insert is safe.

Case 2: you're importing data already validated by another process.

If the data comes from an upstream system that already validated it (another microservice, an ETL), the ORM validator would be redundant. Bulk insert is fine.

Case 3: 1k-10k rows where COPY is overkill.

COPY has a setup cost (raw connection, serialization). For 1k-10k, bulk_insert is simpler and similarly fast.

Case 4: you need standard SQL parameterization.

insert(Model), [...] uses parameterized queries (automatically safe against SQL injection). COPY also does, but the format is different. bulk_insert fits better with the rest of your SQLAlchemy code.

When you should NOT use bulk_insert_mappings

Case 1: the model has a before_insert event with critical logic.

@event.listens_for(Task, 'before_insert')
def assign_owner(mapper, connection, target):
    target.owner_id = get_current_user().id  # depends on context

If this does NOT run with bulk, every bulk-inserted task has owner_id = NULL. Serious bug.

Case 2: @validates does important normalization.

@validates('email')
def normalize_email(self, key, value):
    return value.lower().strip()

Without the validator, emails are inserted with inconsistent casing.

Case 3: complex Python defaults.

class Order(Base):
    order_number: Mapped[str] = mapped_column(
        default=lambda: generate_order_number()  # ID generation logic
    )

Without the Python default, order_number ends up NULL.

Case 4: you need auto-generated IDs immediately.

If you need the IDs for subsequent references (e.g., inserting order_items with order_id), bulk insert without RETURNING doesn't give them to you. Workaround: use RETURNING (slower) or structure the logic differently.

Verifying you don't break functionality

Before switching add_all to bulk_insert, do an audit:

1. Look for events on the model

grep -rn "@event.listens_for(Task" src/
grep -rn "event.listen(Task" src/

2. Look for validators

grep -rn "@validates" src/models/task.py

3. Look for Python defaults

Look at the model definition: any default= that is NOT server_default= may be a Python default.

class Task(Base):
    created_at: Mapped[datetime] = mapped_column(default=lambda: ...)  # ← Python
    updated_at: Mapped[datetime] = mapped_column(server_default=func.now())  # ← SQL OK

4. Regression test

Write a test that uses bulk insert and verifies the expected behavior is preserved:

async def test_bulk_insert_doesnt_break_invariants():
    data = [{"title": "T", "owner_id": None}]  # owner_id not provided

    await session.execute(insert(Task), data)
    await session.commit()

    task = await session.execute(select(Task)).scalar()

    # If your model had an event that sets owner_id, this assert fails with bulk
    assert task.owner_id is not None, "owner_id should be set by event"

If the test fails, you know that switching to bulk breaks something.

Workaround: set defaults manually

If you want bulk speed but need Python defaults, set them manually in the dict:

from datetime import datetime, timezone


def prepare_for_bulk(records):
    """Apply defaults manually."""
    now = datetime.now(timezone.utc)
    return [
        {
            **r,
            "created_at": r.get("created_at", now),
            "owner_id": r.get("owner_id", current_user.id),
        }
        for r in records
    ]


data = prepare_for_bulk(raw_data)
await session.execute(insert(Task), data)

It makes explicit what the event would do implicitly. More code, but clear.

Detailed comparison

Summary: DB-level features work; Python-level features don't. If your logic lives in the DB (triggers, constraints, server_defaults), bulk insert is safe. If it lives in Python (events, validators, defaults), bulk insert hurts you silently.

Pitfalls and common mistakes

1. Assuming default= always works.

default=lambda: now() does not run. If the column is NOT NULL and provides no server_default, the INSERT fails with a constraint error. More obvious. But if the column is nullable, it ends up NULL — a silent bug.

2. Switching to bulk without event tests.

If your codebase has 10 events on the model, switching to bulk silently bypasses them all. Without tests, you won't find out until production.

3. Expecting id from bulk insert without .returning().

# ❌ Python error
result = await session.execute(insert(Task), data)
print(result.inserted_primary_key)  # only works for single insert

For bulk, use an explicit .returning(Task.id).

4. Adding an event to "compensate" for what bulk skips.

Temptation: add an event that "runs before the bulk" — but events are per-row, not per-statement. Bulk doesn't fire them, and there's no way to make it fire them. Solution: set the values in the dict.

5. Worse performance with .returning().

returning() adds overhead. For ~100k rows, it can cancel out much of the bulk gain. If you don't need the IDs, don't use returning.

6. Mixing add() and bulk in the same transaction.

session.add(task1)  # goes through the ORM
await session.execute(insert(Task), [...])  # bypasses the ORM
await session.commit()

It works, but it's confusing. The add() row fires events, the bulk rows don't. Inconsistent behavior.

7. bulk_insert_mappings (1.x style) in 2.0 code.

session.bulk_insert_mappings(Task, mappings) still works in 2.0 but is deprecated. Use await session.execute(insert(Task), mappings).

Exercise: detect what bulk insert skips

Setup: a model with an event, a validator, and defaults.

from sqlalchemy import event
from sqlalchemy.orm import validates


class Task(Base):
    __tablename__ = "tasks"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str] = mapped_column(String(200))
    normalized_title: Mapped[str] = mapped_column(String(200))
    priority: Mapped[int] = mapped_column(default=lambda: 5)
    server_priority: Mapped[int] = mapped_column(server_default="3")
    created_at: Mapped[datetime] = mapped_column(default=lambda: datetime.utcnow())

    @validates('title')
    def normalize(self, key, value):
        # validator that also sets another column
        self.normalized_title = value.lower().strip()
        return value


@event.listens_for(Task, 'before_insert')
def event_logger(mapper, connection, target):
    print(f"[event] inserting: {target.title}")

Step 1: insert with add() and observe.

async def test_add():
    task = Task(title="  Hello  ")
    session.add(task)
    await session.commit()
    print(task.normalized_title)  # → "hello"
    print(task.priority)            # → 5
    print(task.created_at)          # → datetime
    # event print: [event] inserting:   Hello

Step 2: insert with bulk and compare.

async def test_bulk():
    data = [{"title": "  Hello  "}]
    await session.execute(insert(Task), data)
    await session.commit()

    task = await session.execute(select(Task)).scalar()
    print(task.normalized_title)  # → ?
    print(task.priority)           # → ?
    print(task.created_at)         # → ?
    # event print: ?

Step 3: document what gets skipped.

Step 4: decide whether bulk insert is safe for this model.

If you need normalized_title and priority populated, bulk insert is NOT safe.

Step 5: workaround: apply defaults manually.

def prepare_record(raw):
    return {
        "title": raw["title"],
        "normalized_title": raw["title"].lower().strip(),  # Manual
        "priority": raw.get("priority", 5),                 # Manual
        "created_at": raw.get("created_at", datetime.utcnow()),
    }


data = [prepare_record({"title": "  Hello  "})]
await session.execute(insert(Task), data)

normalized_title: "" (NULL if nullable, error if NOT NULL)
priority: NULL (Python default NOT executed, server_default WOULD work)
created_at: NULL
event print: (empty — event not fired)

Summary and next step

What you learned:

• insert(Model), [...] in SQLAlchemy 2.0+ is the modern version of bulk_insert_mappings.
• Skips: ORM events, validators, Python defaults, identity map.
• Keeps: SQL server_default, DB triggers, DB constraints.
• When to use it: models without critical events, 1k-10k rows, COPY is overkill.
• When not to: models with events, validators, Python defaults with no SQL equivalent.
• Workaround: apply defaults manually in the dict before the bulk insert.

Before moving on, you should be able to:

• Audit a model to detect Python events/validators/defaults.
• Decide whether bulk insert is safe for a given model.
• Apply the manual workaround when bulk is desirable but there are Python defaults.
• Tell default= (Python) apart from server_default= (SQL).

In the next capsule we move to a complementary pattern: upserts with ON CONFLICT. How do you insert rows that may already exist, without an error and without a race condition? INSERT ... ON CONFLICT (col) DO UPDATE SET ... is the canonical pattern — atomic, idempotent. You'll learn the exact syntax, the three types of conflict resolution (DO NOTHING, DO UPDATE, partial), and how to combine it with bulk insert.

Resources

1. SQLAlchemy 2.0 — Bulk Insert — official reference.
2. SQLAlchemy 2.0 — Insertmanyvalues optimization — the automatic optimizer.
3. SQLAlchemy ORM — Events — events reference.
4. SQLAlchemy — @validates — validators reference.
5. Migration guide 1.x → 2.0 — includes changes to bulk operations.
6. Mike Bayer (SQLAlchemy author) — Bulk performance — official examples with benchmarks.

Capsule 04 of 08 — Module 7 — SQL Patterns for Production APIs Guide