Loading rows with Postgres COPY through asyncpg

Get the asyncpg connection from the session with (await (await session.connection()).get_raw_connection()).driver_connection and call copy_records_to_table() on it — the rows travel over the binary COPY protocol inside the session's own transaction, typically an order of magnitude faster than executemany. This guide belongs to high-performance bulk inserts and updates.

Quick Answer

SQLAlchemy has no COPY construct, but asyncpg has a fast one, and the session will hand you the connection it is already using.

Before — a million ORM objects, flushed:

from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import PageView


async def load_page_views(session: AsyncSession, rows: list[dict]) -> None:
    session.add_all(PageView(**row) for row in rows)
    await session.commit()

After — binary COPY on the session's connection:

from sqlalchemy.ext.asyncio import AsyncSession

from shop.models import PageView

COLUMNS = ("viewed_at", "customer_id", "path", "duration_ms")


async def load_page_views(session: AsyncSession, rows: list[tuple]) -> int:
    conn = await session.connection()
    raw = await conn.get_raw_connection()
    asyncpg_conn = raw.driver_connection

    status = await asyncpg_conn.copy_records_to_table(
        PageView.__tablename__,
        records=rows,           # an iterable of tuples in COLUMNS order
        columns=COLUMNS,
        schema_name="public",
    )
    await session.commit()
    return int(status.split()[-1])   # "COPY 1000000"

rows is a list of tuples whose values are already the Python types asyncpg's binary codecs expect — datetime with a timezone for timestamptz, int, str. The return value is PostgreSQL's command tag, which carries the row count.

Execution Context & Async Workflow Integration

An AsyncSession does not hold a connection until it needs one. await session.connection() makes it check one out and begin its transaction if it has not already, and returns the AsyncConnection. get_raw_connection() returns the pool's proxy for the physical connection, and driver_connection is the asyncpg Connection object inside it. All three refer to the same socket and the same PostgreSQL transaction.

That shared transaction is what makes the pattern safe. The COPY is just another command on the connection, so rows it loads are invisible to other sessions until session.commit(), and a failure anywhere in the unit of work — the COPY itself, or an ORM flush after it — rolls everything back together. It also means you can mix freely: create an import-batch row through the ORM, COPY a million detail rows that reference it, update a counter, and commit once.

COPY is fast for structural reasons rather than clever ones. An INSERT executed many times, even as a prepared statement, is a protocol exchange per row or per batch: bind, execute, wait for a result. COPY is a single command followed by a stream of rows with no per-row response, and asyncpg's copy_records_to_table encodes those rows in PostgreSQL's binary format on the client, so the server does no text parsing either. The comparison with Core executemany is a large constant factor, and it widens as rows get wider.

What COPY does not do is equally structural. It does not run through SQLAlchemy at all, so Python-side column default= values, @validates methods and ORM events are all bypassed — only server-side defaults, constraints and triggers apply. It has no ON CONFLICT, so a duplicate key aborts the whole COPY and, with it, the transaction. And the ORM does not know the rows exist: objects of that class already in the session are not affected, and nothing is added to the identity map.

Because the COPY runs on the event loop through asyncpg, a very large load does not block other tasks, but it does occupy one pooled connection for its full duration. Long loads belong in a worker process or a dedicated engine rather than in a request handler.

Resolving Warnings, Errors & Common Mistakes

Type mismatches are the most frequent failure, because binary COPY has no server-side casting. A text INSERT of '19.99' into a numeric column works because the server parses the string; binary COPY encodes each value on the client using the codec for the column's type, and a string handed to the numeric codec fails. Normalise rows before loading:

import datetime as dt
import decimal
import json


def to_copy_row(record: dict) -> tuple:
    return (
        record["sku"],
        decimal.Decimal(str(record["price"])),                 # numeric
        dt.datetime.fromisoformat(record["seen_at"]).astimezone(dt.timezone.utc),  # timestamptz
        json.dumps(record.get("attributes", {})),              # jsonb
    )

Timezones deserve extra attention because the failure mode differs between COPY and ordinary queries — fixing timezone-aware datetime errors with asyncpg covers the rules for timestamp versus timestamptz columns.

Advanced: Staging Tables for Upserts and Full Syncs

COPY cannot resolve conflicts, but it can fill a table that has no constraints to conflict with. A temporary staging table turns COPY into the fastest path for upserts too: load the feed into staging, then merge it with a single set-based statement.

import sqlalchemy as sa
from sqlalchemy.ext.asyncio import AsyncSession

STAGING_COLUMNS = ("sku", "price_cents", "stock", "attributes")


async def sync_supplier_feed(session: AsyncSession, rows: list[tuple]) -> dict[str, int]:
    await session.execute(sa.text(
        "CREATE TEMP TABLE staging_products "
        "(sku text, price_cents integer, stock integer, attributes jsonb) ON COMMIT DROP"
    ))

    conn = await session.connection()
    asyncpg_conn = (await conn.get_raw_connection()).driver_connection
    await asyncpg_conn.copy_records_to_table(
        "staging_products", records=rows, columns=STAGING_COLUMNS
    )

    upserted = await session.execute(sa.text(
        "INSERT INTO products (sku, price_cents, stock, attributes) "
        "SELECT sku, price_cents, stock, attributes FROM staging_products "
        "ON CONFLICT (sku) DO UPDATE SET "
        "  price_cents = EXCLUDED.price_cents, "
        "  stock = EXCLUDED.stock, "
        "  attributes = EXCLUDED.attributes "
        "WHERE (products.price_cents, products.stock, products.attributes) "
        "  IS DISTINCT FROM (EXCLUDED.price_cents, EXCLUDED.stock, EXCLUDED.attributes)"
    ))
    retired = await session.execute(sa.text(
        "UPDATE products SET stock = 0 "
        "WHERE supplier_managed AND stock <> 0 "
        "AND NOT EXISTS (SELECT 1 FROM staging_products s WHERE s.sku = products.sku)"
    ))
    await session.commit()          # ON COMMIT DROP removes the staging table
    return {"upserted": upserted.rowcount, "retired": retired.rowcount}

Three details make this robust. Temporary tables are private to the connection and are not WAL-logged, so the COPY into them is cheap and invisible to everyone else. ON COMMIT DROP means a failed run leaves nothing behind once the transaction ends. And the IS DISTINCT FROM guard skips rows that did not change, which keeps an hourly sync of a mostly-unchanged catalogue from rewriting every row and generating dead tuples for autovacuum to clean up.

The temporary table must be created on the same connection that runs the COPY, which the session guarantees as long as everything happens between the first session.execute() and the commit. For the conflict semantics themselves — which columns to update, and what EXCLUDED refers to — see writing Postgres ON CONFLICT DO UPDATE upserts.

Streaming Files and Generators Without Holding Them in Memory

copy_records_to_table accepts any iterable — and, in current asyncpg releases, an asynchronous iterable — so a load does not need every row in memory at once. For a multi-gigabyte export from another system, stream it.

When the source is already a CSV file that PostgreSQL can parse, skip Python row objects entirely and use copy_to_table, which sends the file's bytes and lets the server parse them:

from pathlib import Path

from sqlalchemy.ext.asyncio import AsyncSession


async def load_csv(session: AsyncSession, path: Path) -> str:
    conn = await session.connection()
    asyncpg_conn = (await conn.get_raw_connection()).driver_connection
    with path.open("rb") as source:
        status = await asyncpg_conn.copy_to_table(
            "page_views",
            source=source,
            columns=["viewed_at", "customer_id", "path", "duration_ms"],
            format="csv",
            header=True,
        )
    await session.commit()
    return status

When rows need transformation, an async generator keeps memory flat and lets the transformation read from another async source, such as an HTTP stream or another database:

from collections.abc import AsyncIterator


async def page_view_rows(events: AsyncIterator[dict]) -> AsyncIterator[tuple]:
    async for event in events:
        if event.get("bot"):
            continue
        yield (event["ts"], event["customer_id"], event["path"], event["duration_ms"])


async def load_stream(session, events: AsyncIterator[dict]) -> None:
    conn = await session.connection()
    asyncpg_conn = (await conn.get_raw_connection()).driver_connection
    await asyncpg_conn.copy_records_to_table(
        "page_views",
        records=page_view_rows(events),
        columns=["viewed_at", "customer_id", "path", "duration_ms"],
    )
    await session.commit()

A single enormous COPY is one transaction, which has its own costs: a failure at row nine million discards the first nine million, and the transaction holds back vacuum for its whole duration. For loads that run for many minutes, split the stream into chunks of a few hundred thousand rows and commit each chunk in its own session, recording progress so a restart can resume. The batching trade-offs are the same ones discussed in batch inserting millions of rows with Core execute.

Frequently Asked Questions

Related

• High-Performance Bulk Inserts and Updates — The parent guide: choosing a bulk write path.
• Benchmarking Core executemany bulk insert performance — Where executemany is fast enough, and where it is not.
• Bulk upserting rows with INSERT ON CONFLICT — The merge step after staging.
• Using ORM-enabled UPDATE and DELETE statements — Set-based updates without loading objects.