The 4 approaches with real benchmarks

"COPY is faster" in the abstract doesn't convince anybody. 47 seconds vs 0.8 seconds does. This capsule shows you the 4 bulk-insert approaches with reproducible benchmarks. You're going to run each one with the same data (100,000 rows), measure the time, and see the difference with your own eyes. Afterward you're not going to forget when to choose each one.

The 4 approaches are: a loop INSERT, executemany, bulk_insert_mappings, and COPY. Each one has its place — none is always the right one. The idea of this capsule is that you measure it yourself and develop intuition.

The benchmark setup: a simple table, 100k rows, the same hardware, the same connection. The variable: the approach. We measure time.perf_counter() before and after.

---

The benchmark setup

# benchmark_setup.py
import asyncio
from sqlalchemy import String, Integer, Numeric
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column


DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost/test"


class Base(DeclarativeBase):
    pass


class Task(Base):
    __tablename__ = "tasks_bench"

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    title: Mapped[str] = mapped_column(String(200))
    status: Mapped[str] = mapped_column(String(50))
    priority: Mapped[int] = mapped_column(Integer)


async def setup():
    engine = create_async_engine(DATABASE_URL)
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
    await engine.dispose()


asyncio.run(setup())

Data generation:

def generate_data(n: int) -> list[dict]:
    return [
        {
            "title": f"Task {i}",
            "status": "pending" if i % 3 != 0 else "completed",
            "priority": i % 5 + 1,
        }
        for i in range(n)
    ]


N = 100_000
data = generate_data(N)

---

Approach 1: a loop INSERT

import time
import asyncio


async def benchmark_loop():
    engine = create_async_engine(DATABASE_URL)
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    # Truncate first
    async with engine.begin() as conn:
        await conn.execute(text("TRUNCATE tasks_bench"))

    start = time.perf_counter()

    async with SessionLocal() as session:
        for row in data:
            task = Task(**row)
            session.add(task)
        await session.commit()

    elapsed = time.perf_counter() - start
    print(f"Loop INSERT (n={N}): {elapsed:.2f}s")


asyncio.run(benchmark_loop())

A typical result:

Loop INSERT (n=100000): 47.32s

Throughput: ~2,100 inserts/second. Each INSERT has the overhead of the SQLAlchemy ORM (event listeners, the identity map, etc.) plus a network round-trip if the DB isn't local.

---

Approach 2: executemany

session.execute(insert(Model), [...]) runs multiple INSERTs in a single call to the driver. PostgreSQL receives them in a batch, reducing round-trips.

from sqlalchemy import insert


async def benchmark_executemany():
    engine = create_async_engine(DATABASE_URL)
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    async with engine.begin() as conn:
        await conn.execute(text("TRUNCATE tasks_bench"))

    start = time.perf_counter()

    async with SessionLocal() as session:
        await session.execute(insert(Task), data)
        await session.commit()

    elapsed = time.perf_counter() - start
    print(f"executemany (n={N}): {elapsed:.2f}s")


asyncio.run(benchmark_executemany())

A typical result:

executemany (n=100000): 12.18s

Throughput: ~8,200 inserts/second. ~4x faster than the loop. It still goes through the ORM (processing each row), but without the overhead of the identity map or per-row event listeners.

Note: in recent versions of SQLAlchemy 2.0+, this approach uses the "insertmanyvalues" optimization automatically, grouping multiple rows into fewer SQL statements.

---

Approach 3: bulk_insert_mappings

session.bulk_insert_mappings(Model, [...]) bypasses the ORM completely. It doesn't create Task instances — it just builds parameterized SQL and executes it.

async def benchmark_bulk_insert():
    engine = create_async_engine(DATABASE_URL)
    SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

    async with engine.begin() as conn:
        await conn.execute(text("TRUNCATE tasks_bench"))

    start = time.perf_counter()

    async with SessionLocal() as session:
        # SQLAlchemy 2.0 syntax
        await session.execute(insert(Task), data)
        # The equivalent of "bulk_insert_mappings" in earlier versions
        # In 2.0+, insert(Table) with a list is the canonical form
        await session.commit()

    elapsed = time.perf_counter() - start
    print(f"bulk_insert (n={N}): {elapsed:.2f}s")


asyncio.run(benchmark_bulk_insert())

A typical result:

bulk_insert (n=100000): 4.27s

Throughput: ~23,400 inserts/second. ~3x faster than executemany. No event listeners, no identity map, no instance creation. Just SQL.

Critical limitations:

• It doesn't fire before_insert/after_insert events.
• It doesn't fire @validates decorators.
• It doesn't apply Python column defaults (default=lambda: ...) — only SQL defaults (server_default=).
• It doesn't return autogenerated PKs (you need to re-fetch).

If your model depends on any of these for correctness, do NOT use bulk_insert_mappings.

---

Approach 4: COPY with asyncpg

COPY FROM STDIN is PostgreSQL's native bulk loader. It bypasses per-row SQL parsing — it receives data in binary or CSV format and writes it directly to the heap.

import asyncpg


async def benchmark_copy():
    # asyncpg directly (no SQLAlchemy)
    conn = await asyncpg.connect(
        "postgresql://postgres:postgres@localhost/test"
    )

    await conn.execute("TRUNCATE tasks_bench")

    start = time.perf_counter()

    # copy_records_to_table accepts a list of tuples
    records = [(row["title"], row["status"], row["priority"]) for row in data]

    await conn.copy_records_to_table(
        "tasks_bench",
        records=records,
        columns=["title", "status", "priority"],
    )

    elapsed = time.perf_counter() - start
    print(f"COPY (n={N}): {elapsed:.2f}s")

    await conn.close()


asyncio.run(benchmark_copy())

A typical result:

COPY (n=100000): 0.83s

Throughput: ~120,000 inserts/second. 60x faster than the loop, 14x faster than executemany.

Why is it so fast? PostgreSQL receives the data in an efficient binary format, with no per-row SQL parsing, no constraint checks in a verbose format. It's literally "copy this to the heap, then rebuild the indexes".

---

The comparison table

Approach	Time	Throughput	ORM events	Limitations
Loop INSERT	47.32s	2.1k/s	Yes (all)	Slow, doesn't scale
executemany	12.18s	8.2k/s	Yes (all)	Acceptable up to ~10k rows
bulk_insert	4.27s	23.4k/s	No	Skips events, validators, Python defaults
COPY	0.83s	120k/s	No	Doesn't support ON CONFLICT directly, requires a temp table for upserts

From the end client's perspective:

• 100 rows: any of them. A <100ms difference is invisible.
• 1,000 rows: executemany is enough.
• 10,000 rows: bulk_insert is preferable (4s vs 12s = 3x better).
• 100,000+ rows: COPY is mandatory (0.8s vs 4s = another 5x improvement).

---

The curve isn't linear

As N increases, the curve doesn't scale linearly:

N	Loop	executemany	bulk_insert	COPY
1k	0.5s	0.15s	0.06s	0.02s
10k	4.5s	1.3s	0.5s	0.1s
100k	47s	12s	4.3s	0.8s
1M	~480s	~115s	~42s	~7s
10M	OOM or a timeout	~1100s	~410s	~62s

A loop at 10M rows is already a minutes-long operation. With a loop, 100M rows are hours → days. With COPY, 100M rows are ~10 minutes.

The lesson: the right approach depends on the current size and the expected growth. If your app is going to import 100k rows today and 10M in 6 months, write COPY from day 1.

---

Traps and common mistakes

1. Comparing approaches without a truncate.

If you don't TRUNCATE before each benchmark, the previous run's inserts affect the next one (through bloat, through triggers, through checks). Truncating first guarantees identical conditions.

2. Running in psql and comparing against Python code.

psql \copy is local, with no network or Python overhead. Comparing it against Python directly is unfair. Compare approaches within Python.

3. Forgetting await session.commit().

Without the commit, the rows don't get persisted — the benchmark looks "super fast" but it didn't do anything real.

4. Using INSERT ... RETURNING when you don't need the IDs.

RETURNING * requires PostgreSQL to return all the inserted columns. For bulk, this is a large overhead. If you don't need the IDs, don't use RETURNING.

5. Measuring from inside a test framework with a slow teardown.

If you run the benchmark inside pytest with a teardown that does cleanup, the timings can be inflated. Run benchmarks in a standalone script.

6. Comparing approaches with different indexes.

If your table has 5 indexes, each INSERT updates 5 indexes. Every approach pays that cost. But if you compare a table with indexes against a table without, the numbers aren't comparable. The same schema in all the benchmarks.

7. Assuming your hardware gives the same numbers.

47s on my laptop can be 80s in your container, 25s on your bare-metal server. What matters is the ratio between approaches (a 60x difference) — that one is consistent.

8. Not considering concurrency.

These benchmarks are single-connection. In real production, multiple concurrent requests can saturate the DB. COPY consumes more CPU/IO than executemany — if you have 10 concurrent imports, executemany can be better because of less contention.

---

Exercise: reproduce the benchmarks

Setup: PostgreSQL locally (via Docker or native), Python 3.12+, pip install sqlalchemy[asyncio] asyncpg.

Step 1: create the script benchmark.py with the 4 approaches.

# benchmark.py
import asyncio
import time

# ... copy the functions from above

async def main():
    print(f"Benchmarking with N = {N}")
    await benchmark_loop()
    await benchmark_executemany()
    await benchmark_bulk_insert()
    await benchmark_copy()


if __name__ == "__main__":
    asyncio.run(main())

Step 2: run it.

python benchmark.py

Step 3: document the results.

Approach	Time	Throughput
Loop	?	?
executemany	?	?
bulk_insert	?	?
COPY	?	?

Step 4: repeat with different N (1k, 10k, 1M) and plot the curve.

Step 5: experiment.

• What happens if you comment out await session.commit() in one of them?
• What happens if you add an index to the table? How does each approach change?
• What happens if the table has a PL/pgSQL BEFORE INSERT trigger? Which approaches fire it?

See discussion

Step 4 — the curve with different N:

The curve shows the difference becomes more pronounced with a large N. For N=100, the 4 approaches are all in the millisecond range — an invisible difference. For N=1M, the difference is minutes.

Step 5 — the experiments:

With no commit: the data doesn't persist. The function "finishes" but the table is empty. If you compare benchmarks "with no commit", the numbers are false.

With indexes: each INSERT updates each index. The loop and executemany suffer more (each row pays the cost). bulk_insert and COPY do too but less (PostgreSQL can optimize batch updates). For massive imports into a table with many indexes, consider dropping the indexes → importing → recreating the indexes.

With a PL/pgSQL trigger: PostgreSQL triggers (not SQLAlchemy ones) DO fire in ALL the approaches including COPY. What gets skipped with bulk_insert/COPY are the Python ORM events, not the DB triggers.

The key lessons:

1. The difference is real and reproducible.
2. The ratio is what matters (60x), not the absolute number.
3. The trade-offs are different: COPY is fastest but loses the Python events; bulk_insert is the middle balance.
4. DB triggers (PL/pgSQL) do work with all of them.

---

Summary and next step

What you learned:

• 4 approaches with real benchmarks: 47s → 12s → 4.3s → 0.8s for 100k rows.
• A loop INSERT: simple, enough for <100 rows.
• executemany: good up to ~10k rows, keeps the events.
• bulk_insert (insert(Model), list): fast but skips the Python events.
• COPY: 60x faster than the loop, requires asyncpg directly (not the SQLAlchemy ORM).
• A non-linear curve: the differences amplify with a large N.
• Specific trade-offs: ORM events, Python defaults, RETURNING IDs.

Before moving on, you should be able to:

• Reproduce the benchmarks in your setup.
• Decide the right approach based on N and the constraints (are the events needed?).
• Explain why COPY is so fast (the binary format, no per-row parse).
• Justify the decision matrix by size in code review.

In the next capsule we go deep on the most powerful one: COPY with asyncpg. You're going to learn the complete pattern — copy_records_to_table vs copy_to_table, serializing special types (datetime, JSON, NULL), the binary format for extreme cases, and error handling. It's the pattern you use when bulk_insert isn't enough.

---

Resources

1. PostgreSQL Docs — COPY — the official reference.
2. asyncpg — copy_records_to_table — the reference.
3. SQLAlchemy 2.0 — Insertmanyvalues — the executemany optimization in 2.0+.
4. Brandur Leach — Postgres bulk insert benchmarks — in-depth benchmarks.
5. Citus Data — Bulk loading benchmarks — a detailed comparison.
6. pg_bulkload — an extension for extreme cases (>100M rows).
7. Tom Augspurger — pandas to_sql performance — the data engineering perspective.

---

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