Batch Inserting Millions of Rows with SQLAlchemy Core

To batch insert millions of rows with SQLAlchemy 2.0 Core, pass a list of dicts to await conn.execute(insert(table), chunk) inside engine.begin() — the engine delegates to the driver's native executemany path, which on asyncpg uses binary protocol encoding and a single server-side prepared statement for the entire batch. This page is the memory-safe, production-scale companion to the parent guide on high-performance bulk inserts and updates, which covers upsert, RETURNING, and synchronize_session in full context.

Quick Answer

# Legacy 1.4 — session.bulk_insert_mappings (removed in 2.0)
session.bulk_insert_mappings(Order, list_of_dicts)

# Modern 2.0 — Core async bulk insert with explicit chunking
import asyncio
from itertools import islice
from sqlalchemy import insert, Table, MetaData, Column, Integer, String, Numeric
from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine

metadata = MetaData()
orders_table = Table(
    "orders", metadata,
    Column("id", Integer, primary_key=True),
    Column("customer_id", Integer, nullable=False),
    Column("amount", Numeric(12, 4), nullable=False),
    Column("status", String(32), nullable=False),
)

async def batch_insert(engine: AsyncEngine, rows: list[dict], chunk_size: int = 2_000) -> None:
    stmt = insert(orders_table)
    it = iter(rows)
    async with engine.begin() as conn:
        while chunk := list(islice(it, chunk_size)):
            await conn.execute(stmt, chunk)

engine.begin() wraps each block in an auto-committing transaction — no conn.commit() call required. To commit per-chunk (for very large datasets where a single transaction would exhaust WAL), move engine.begin() inside the loop. The Core path here is deliberate: going through insert(table) rather than the ORM sidesteps the unit-of-work machinery described in Core vs ORM architecture decisions, which is the single largest throughput lever for ingestion workloads.

Execution Context & Async Workflow Integration

How executemany Maps to the Driver

When conn.execute(stmt, list_of_dicts) receives a sequence, SQLAlchemy compiles the insert statement once and hands the parameter list to the underlying DBAPI's executemany() method. On asyncpg this translates to:

1. A single Parse message — the server prepares the statement once.
2. N Bind + Execute messages, each carrying one row's parameters in efficient binary format.
3. A single Sync message to flush and confirm.

No SQL string is re-parsed per row. The binary encoding avoids text-to-value coercion overhead on both the Python and PostgreSQL sides. The net effect is that asyncpg sustains 60,000–100,000 rows per second on typical cloud hardware for narrow tables, compared to 8,000–15,000 for ORM add_all() with the same row shape — a gap quantified strategy-by-strategy in benchmarking Core executemany bulk insert performance.

One caveat with the "single server-side prepared statement" behaviour: if your ingestion connections route through PgBouncer in transaction pooling mode, asyncpg's prepared statements break across pooled backends. See handling asyncpg prepared statement errors with PgBouncer for the statement_cache_size=0 fix before pointing a bulk loader at a pooled endpoint.

Generator-Based Streaming for Multi-Million Row Sources

Loading an entire CSV or upstream API response into a Python list before inserting causes MemoryError at scale. Use a generator that yields row dicts on demand:

import csv
from typing import Iterator

def stream_orders_from_csv(filepath: str) -> Iterator[dict]:
    """Yield one dict per CSV row — never holds more than one row in memory."""
    with open(filepath, encoding="utf-8", newline="") as fh:
        for row in csv.DictReader(fh):
            yield {
                "customer_id": int(row["customer_id"]),
                "amount": row["amount"],       # Decimal-compatible string
                "status": row.get("status", "pending"),
            }

Pair this with the chunking pattern from the Quick Answer section. islice materializes only chunk_size dicts at a time, keeping Python heap usage flat regardless of source size. This is the write-side mirror of streaming large result sets with yield_per: both patterns keep memory bounded by processing the dataset in fixed-size windows rather than materializing it whole.

Per-Chunk Transaction Boundaries

For data sets larger than ~500,000 rows, a single transaction is impractical because:

• PostgreSQL's WAL accumulates all changes before flushing — a single 10 M row transaction can generate 10–40 GB of WAL.
• Autovacuum cannot reclaim dead tuples inside an open transaction.
• Any failure forces a full rollback of all previously inserted rows.

Commit per chunk and track a resumption checkpoint to support safe restarts:

import asyncio
from itertools import islice
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy import Table, insert

async def resumable_bulk_insert(
    engine: AsyncEngine,
    table: Table,
    data_stream,
    chunk_size: int = 2_000,
    start_offset: int = 0,
) -> int:
    """Returns total rows inserted; supports restart from start_offset."""
    stmt = insert(table)
    total = 0
    it = islice(data_stream, start_offset, None)   # skip already-inserted rows

    while chunk := list(islice(it, chunk_size)):
        async with engine.begin() as conn:          # one transaction per chunk
            await conn.execute(stmt, chunk)
        total += len(chunk)
    return total

Async Concurrency: Parallel Chunk Workers

For I/O-bound pipelines with multiple CPU-bound data sources, run chunk insertions concurrently using asyncio.gather(). Use a semaphore to cap database connections:

import asyncio
from sqlalchemy.ext.asyncio import AsyncEngine
from sqlalchemy import Table, insert

async def parallel_bulk_insert(
    engine: AsyncEngine,
    table: Table,
    chunks: list[list[dict]],
    max_concurrency: int = 5,
) -> None:
    sem = asyncio.Semaphore(max_concurrency)
    stmt = insert(table)

    async def insert_chunk(chunk: list[dict]) -> None:
        async with sem:
            async with engine.begin() as conn:
                await conn.execute(stmt, chunk)

    await asyncio.gather(*[insert_chunk(c) for c in chunks])

Keep max_concurrency at or below pool_size to prevent connection starvation. Sizing that pool correctly is its own topic — configuring async engines and connection pools covers the pool_size/max_overflow math for concurrent writers. Monitor pg_stat_activity for wait_event = 'ClientRead' spikes that indicate connection queue pressure.

Resolving Warnings, Errors & Common Mistakes

Advanced Bulk Insert Optimization

insertmanyvalues vs Standard executemany

SQLAlchemy 2.0 introduced insertmanyvalues: when the backend supports RETURNING (PostgreSQL, SQLite ≥ 3.35), and the statement includes .returning(), SQLAlchemy rewrites the insert as batched multi-value INSERT ... VALUES (...), (...) RETURNING ... grouped in pages of up to insertmanyvalues_page_size rows (default 1000).

This is measurably faster than standard executemany for workloads that need to capture generated PKs, because it reduces round-trips from N (one per row) to ceil(N / page_size). For workloads that do not need RETURNING, standard executemany with asyncpg is comparably fast.

from sqlalchemy import insert
from sqlalchemy.ext.asyncio import AsyncEngine

async def insert_with_returning(
    engine: AsyncEngine, rows: list[dict]
) -> list[int]:
    stmt = (
        insert(orders_table)
        .returning(orders_table.c.id)
        .execution_options(insertmanyvalues_page_size=2_000)
    )
    async with engine.begin() as conn:
        result = await conn.execute(stmt, rows)
        return [row.id for row in result]

asyncpg COPY for Maximum Throughput

When throughput requirements exceed 200,000 rows per second, abandon the SQL insert path entirely and use asyncpg's binary COPY protocol. Access the raw connection via conn.get_raw_connection():

import asyncpg
from sqlalchemy.ext.asyncio import AsyncEngine

async def copy_orders(engine: AsyncEngine, records: list[tuple]) -> None:
    """
    records: list of (customer_id: int, amount: Decimal, status: str)
    Column 'id' is omitted — PostgreSQL generates it via SERIAL/GENERATED.
    """
    async with engine.connect() as conn:
        raw: asyncpg.Connection = await conn.get_raw_connection()
        await raw.copy_records_to_table(
            "orders",
            records=records,
            columns=["customer_id", "amount", "status"],
            timeout=120.0,
        )
        await conn.commit()

COPY bypasses row-by-row statement processing and streams data in the PostgreSQL wire binary format. It is 3–6× faster than insertmanyvalues for large batches. Constraint verification still occurs at commit — validate FK integrity before calling COPY to avoid full-batch rollbacks.

Advanced Batch Insert Patterns

Backpressure and Rate Limiting in Long-Running Pipelines

In production ingestion pipelines, inserting as fast as the database will accept data without any pacing causes two problems: it exhausts connection pool slots from concurrent workers, and it spikes WAL generation faster than PostgreSQL's background writer can flush to disk. Adding a concurrency semaphore provides implicit backpressure:

import asyncio
from sqlalchemy import insert, Table
from sqlalchemy.ext.asyncio import AsyncEngine

async def paced_bulk_insert(
    engine: AsyncEngine,
    table: Table,
    chunks: list[list[dict]],
    max_concurrent: int = 4,
) -> None:
    sem = asyncio.Semaphore(max_concurrent)
    stmt = insert(table)

    async def insert_one(chunk: list[dict]) -> None:
        async with sem:
            async with engine.begin() as conn:
                await conn.execute(stmt, chunk)

    await asyncio.gather(*[insert_one(c) for c in chunks])

Set max_concurrent to match pool_size on the engine. Running more concurrent insert coroutines than available pool slots causes them to queue inside SQLAlchemy's connection pool, which is fine for throughput but creates long pool-wait latencies that mask the actual database-side bottleneck.

Validating Row Schema Before Insert

Silent type coercion failures are the hardest bulk insert bugs to diagnose: PostgreSQL may accept a string "123" for an INTEGER column through the text protocol, but asyncpg's binary protocol will reject it with DataError: invalid input for query argument. Pre-validate dict keys and types before passing to execute:

from sqlalchemy import Table
from typing import Any

def validate_rows(table: Table, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Ensure all rows have only known column keys and no None in NOT NULL columns."""
    valid_keys = {col.name for col in table.c}
    nullable = {col.name for col in table.c if col.nullable}
    errors = []
    for i, row in enumerate(rows):
        unknown = set(row) - valid_keys
        if unknown:
            errors.append(f"Row {i}: unknown keys {unknown}")
        for col in table.c:
            if not col.nullable and col.name != "id" and row.get(col.name) is None:
                errors.append(f"Row {i}: NULL in NOT NULL column '{col.name}'")
    if errors:
        raise ValueError(f"Row validation failed:\n" + "\n".join(errors[:10]))
    return rows

Run this validation on a sample of the first chunk during development, then disable it or move it to a separate offline validation step for production throughput.

Frequently Asked Questions

Related

• High-Performance Bulk Inserts and Updates — parent guide covering upsert, RETURNING, synchronize_session, and asyncpg COPY in full context.
• Benchmarking Core Executemany Bulk Insert Performance — measuring and comparing ORM, Core executemany, and COPY with real throughput numbers.
• Configuring Async Engines and Connection Pools — pool sizing and connect_args tuning that directly affects bulk insert throughput.
• Streaming Large Result Sets with yield_per — the read-side complement to bulk inserts for processing large datasets incrementally.