`COPY` in depth with asyncpg

COPY is PostgreSQL's native bulk loader — the fastest approach when you have to import large volumes. But using it well from Python requires understanding several details: how to serialize special types (datetime, NULL, JSON), how to handle transactional errors (it's atomic — it fails all or nothing), how to choose between copy_records_to_table and copy_to_table, and when the binary format beats CSV.

In this capsule you're going to learn the complete COPY pattern with asyncpg. By the end you'll have the canonical snippets to reuse in any endpoint that needs a bulk import.

The 3 COPY variants in asyncpg

asyncpg exposes 3 functions for COPY:

1. copy_records_to_table — the simplest

import asyncpg


async def import_simple():
    conn = await asyncpg.connect("postgresql://...")

    records = [
        ("Task 1", "pending", 5),
        ("Task 2", "completed", 3),
        ("Task 3", "pending", 1),
    ]

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

    await conn.close()

When to use it: most cases. It takes a list of tuples (or an iterable) and inserts them. It handles serialization automatically for common types (str, int, datetime, etc.).

Advantages:

• A simple, intuitive API.
• It handles Python types directly.
• A fast path for in-memory datasets.

Limitations:

• It requires the data in memory (the complete list).
• No streaming from files directly.

2. copy_from_query — from a query result

async def copy_from_query():
    conn = await asyncpg.connect("...")

    # Copy the result of a SELECT to another table
    await conn.copy_from_query(
        "SELECT id, title, status FROM source_tasks WHERE archived = TRUE",
        output="archived_tasks_csv.csv",
        format="csv",
        header=True,
    )

    await conn.close()

When to use it: when you want to export a query's result to a CSV file.

3. copy_to_table — from a file or a stream

async def copy_from_file():
    conn = await asyncpg.connect("...")

    with open("/path/to/data.csv", "r") as f:
        await conn.copy_to_table(
            "tasks",
            source=f,
            format="csv",
            header=True,
            columns=["title", "status", "priority"],
        )

    await conn.close()

When to use it: when you already have a CSV file on disk. It allows streaming without loading everything into memory.

For this capsule, the focus is copy_records_to_table, the one most used in API endpoints.

The canonical implementation with SQLAlchemy 2.0 async

copy_records_to_table is an asyncpg API. SQLAlchemy 2.0 async with asyncpg lets you access the raw connection:

from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession

router = APIRouter()


@router.post("/tasks/bulk")
async def bulk_import_tasks(
    tasks_data: list[TaskCreate],
    db: AsyncSession = Depends(get_db),
):
    # Convert the Pydantic models to tuples
    records = [
        (t.title, t.status, t.priority)
        for t in tasks_data
    ]

    # Access the raw asyncpg connection from SQLAlchemy
    raw_conn = await db.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    # asyncpg_conn is the underlying asyncpg connection
    raw = asyncpg_conn.driver_connection

    await raw.copy_records_to_table(
        "tasks",
        records=records,
        columns=["title", "status", "priority"],
    )

    await db.commit()

    return {"imported": len(records)}

Access to the raw connection can vary by SQLAlchemy version. A cleaner alternative pattern: have a helper:

# helpers.py
from sqlalchemy.ext.asyncio import AsyncSession


async def get_asyncpg_connection(session: AsyncSession):
    """Get the raw asyncpg connection from a SQLAlchemy session."""
    raw_conn = await session.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    return asyncpg_conn.driver_connection

Type serialization

copy_records_to_table handles common types automatically, but some require care.

Datetime with a timezone

from datetime import datetime, timezone

records = [
    (
        "Task 1",
        datetime(2026, 5, 8, 14, 30, tzinfo=timezone.utc),  # ✅
    ),
    (
        "Task 2",
        datetime.now(),  # ❌ no timezone — an error if the column is TIMESTAMPTZ
    ),
]

PostgreSQL's TIMESTAMPTZ requires timezone-aware values. asyncpg raises an error if you pass a naive datetime to a TIMESTAMPTZ.

NULL

None maps to NULL automatically:

records = [
    ("Task 1", "pending", None),  # priority = NULL
    ("Task 2", None, 5),          # status = NULL
]

JSON / JSONB

For JSON/JSONB columns, you pass a dict directly:

records = [
    ("Task 1", {"meta": "data", "tags": ["urgent"]}),
]

await conn.copy_records_to_table(
    "tasks",
    records=records,
    columns=["title", "metadata"],
    # asyncpg serializes the dict to JSON automatically if the column is JSONB
)

Arrays

Lists map to PostgreSQL arrays:

records = [
    ("Task 1", ["urgent", "bug"]),  # tags TEXT[]
    ("Task 2", [1, 2, 3]),           # numbers INTEGER[]
]

Decimal

For NUMERIC columns, you pass a Decimal:

from decimal import Decimal

records = [
    ("Order 1", Decimal("99.99")),
    ("Order 2", Decimal("150.00")),
]

Bytes

For BYTEA, you pass bytes:

records = [
    ("file1.bin", b"\x00\x01\x02\x03"),
]

Custom types

If your table has a custom enum type or a composite type, asyncpg may require registering the type:

# If you have ENUM TYPE 'task_status' AS ENUM ('pending', 'completed')
await conn.set_type_codec(
    'task_status',
    encoder=str,
    decoder=str,
    schema='public',
)

Performance: an in-memory dataset vs streaming

In memory (the typical case)

records = [
    (t.title, t.status, t.priority)
    for t in tasks_data  # a list in memory
]
await conn.copy_records_to_table("tasks", records=records, columns=[...])

It works perfectly up to ~1M rows (depending on available RAM). 1M rows × 100 bytes = 100MB — manageable.

Streaming from a generator

If you have more than 1M rows, streaming is better:

def generate_records():
    """A generator that yields records one by one."""
    with open("huge_file.csv") as f:
        reader = csv.DictReader(f)
        for row in reader:
            yield (row["title"], row["status"], int(row["priority"]))


# asyncpg accepts iterables, not necessarily lists
records_iter = generate_records()
await conn.copy_records_to_table("tasks", records=records_iter, columns=[...])

asyncpg consumes the iterable in a streaming way — it doesn't load everything into memory. Critical for large datasets.

Streaming from a CSV file directly

If the file is already in a COPY-compatible format:

async def import_from_csv(file_path: str, conn):
    with open(file_path, "rb") as f:  # binary mode!
        await conn.copy_to_table(
            "tasks",
            source=f,
            format="csv",
            header=True,
            columns=["title", "status", "priority"],
        )

Note the "rb" (binary mode). asyncpg needs bytes for efficient streaming.

Error handling

COPY is transactional: one invalid row invalidates the whole batch.

records = [
    ("Task 1", "pending", 5),
    ("Task 2", "invalid_status_too_long_for_column", 3),  # ❌ a string > 50 chars
    ("Task 3", "completed", 1),
]

try:
    await conn.copy_records_to_table("tasks", records=records, columns=[...])
except asyncpg.exceptions.PostgresError as e:
    print(f"COPY failed: {e}")
    # NO row was inserted — an automatic rollback

If the COPY fails, no row gets persisted. PostgreSQL rolls back the whole batch.

The pattern: validate first

To avoid failures in the COPY, validate before passing to the COPY:

async def bulk_import_with_validation(
    raw_data: list[dict],
    conn: asyncpg.Connection,
):
    valid_records = []
    errors = []

    for i, row in enumerate(raw_data):
        try:
            # Validate with Pydantic
            validated = TaskCreate(**row)
            valid_records.append((
                validated.title,
                validated.status,
                validated.priority,
            ))
        except ValidationError as e:
            errors.append({"index": i, "errors": e.errors()})

    # If there are errors, a decision: skip or fail-all
    if errors:
        # Option 1: fail-all (don't import partially)
        return {"imported": 0, "errors": errors, "skipped": True}

        # Option 2: import what's valid
        # (continues below)

    # Import the valid ones
    if valid_records:
        await conn.copy_records_to_table(
            "tasks",
            records=valid_records,
            columns=["title", "status", "priority"],
        )

    return {
        "imported": len(valid_records),
        "errors": errors,
        "skipped": len(errors),
    }

The pattern: COPY into a temporary table + filter

For cases where the per-row validation has to use DB lookups (e.g. foreign keys):

async def import_with_temp_table(records, conn):
    # 1. Create a temp table with the same structure + an error column
    await conn.execute("""
        CREATE TEMP TABLE tmp_tasks_import (
            title TEXT,
            status TEXT,
            priority INT,
            valid BOOL DEFAULT TRUE,
            error TEXT
        ) ON COMMIT DROP
    """)

    # 2. COPY into the temp table — it does NOT validate foreign keys or strict constraints
    await conn.copy_records_to_table(
        "tmp_tasks_import",
        records=records,
        columns=["title", "status", "priority"],
    )

    # 3. Validate in SQL (faster than iterating in Python)
    await conn.execute("""
        UPDATE tmp_tasks_import SET valid = FALSE, error = 'invalid status'
        WHERE status NOT IN ('pending', 'completed', 'archived')
    """)
    await conn.execute("""
        UPDATE tmp_tasks_import SET valid = FALSE, error = 'priority out of range'
        WHERE priority < 1 OR priority > 5
    """)

    # 4. INSERT only the valid ones into the real table
    result = await conn.fetch("""
        INSERT INTO tasks (title, status, priority)
        SELECT title, status, priority FROM tmp_tasks_import WHERE valid = TRUE
        RETURNING id
    """)

    # 5. Report the errors
    errors = await conn.fetch("""
        SELECT title, error FROM tmp_tasks_import WHERE NOT valid
    """)

    return {
        "imported": len(result),
        "errors": [dict(e) for e in errors],
    }

A powerful pattern: COPY does an efficient bulk insert into the temp table; SQL does fast vectorized validation; the INSERT with a WHERE filters only the valid ones.

The binary format vs CSV

asyncpg internally uses the binary format, but copy_to_table with format="csv" lets you use CSV files directly.

For typical cases (data in memory from Python), the binary default is the right one. Only if you have a CSV in a file, use copy_to_table with format="csv".

A real case: the POST /tasks/bulk endpoint

from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status, Header
from pydantic import BaseModel, ValidationError
from sqlalchemy.ext.asyncio import AsyncSession
import time


router = APIRouter()


class TaskCreate(BaseModel):
    title: str
    status: str = "pending"
    priority: int = 1


class BulkImportResponse(BaseModel):
    imported: int
    errors: list[dict]
    elapsed_ms: int


@router.post("/tasks/bulk", response_model=BulkImportResponse)
async def bulk_import_tasks(
    tasks: list[TaskCreate],
    db: AsyncSession = Depends(get_db),
) -> BulkImportResponse:
    if len(tasks) > 50_000:
        raise HTTPException(
            status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail="Max 50,000 tasks per bulk import"
        )

    start = time.perf_counter()

    # Convert to tuples
    records = [
        (t.title, t.status, t.priority)
        for t in tasks
    ]

    # Get the raw connection
    raw_conn = await db.connection()
    asyncpg_conn = await raw_conn.get_raw_connection()
    pg_conn = asyncpg_conn.driver_connection

    try:
        await pg_conn.copy_records_to_table(
            "tasks",
            records=records,
            columns=["title", "status", "priority"],
        )
        await db.commit()
    except Exception as e:
        await db.rollback()
        raise HTTPException(500, f"Import failed: {e}")

    elapsed_ms = int((time.perf_counter() - start) * 1000)

    return BulkImportResponse(
        imported=len(records),
        errors=[],
        elapsed_ms=elapsed_ms,
    )

Test:

curl -X POST http://localhost:8000/tasks/bulk \
  -H "Content-Type: application/json" \
  -d '[
    {"title": "T1", "status": "pending", "priority": 5},
    {"title": "T2", "status": "pending", "priority": 3}
  ]'
# {"imported": 2, "errors": [], "elapsed_ms": 8}

For 10k tasks:

# Generate the payload
python -c "import json; print(json.dumps([{'title': f'Task {i}', 'status': 'pending', 'priority': i % 5 + 1} for i in range(10000)]))" > tasks.json

curl -X POST http://localhost:8000/tasks/bulk \
  -H "Content-Type: application/json" \
  -d @tasks.json
# {"imported": 10000, "errors": [], "elapsed_ms": 145}

145ms for 10k rows. Comparable to SQLAlchemy's bulk_insert at this size, it wins by a lot at >100k.

Traps and common mistakes

1. Not committing after the COPY.

copy_records_to_table runs inside the transaction. If you don't do db.commit(), the data doesn't persist. Forgetting it is a common bug.

2. Mixing COPY with session.add() in the same transaction.

# An anti-pattern
session.add(task1)
await pg_conn.copy_records_to_table(...)  # affects different rows
await session.commit()

It works but it's confusing. If you need bulk + individual operations, do them in separate transactions or everything via COPY.

3. Assuming asyncpg knows all your custom types.

For custom enums, composite types, etc., register a codec with set_type_codec or pass the values as plain strings.

4. Passing large lists to the endpoint with no limit.

The client can send 10M rows and your endpoint tries to process everything. Set a limit (max_items) in the Pydantic validation or an early check.

5. Forgetting the rollback on an error.

Without await db.rollback() after the exception, the session stays in a broken state. Subsequent queries fail with "session in failed state".

6. Using copy_records_to_table with each row in a separate chunk.

# ❌
for chunk in chunks(records, 100):
    await conn.copy_records_to_table(..., records=chunk)

# ✅ Pass everything at once
await conn.copy_records_to_table(..., records=records)

asyncpg handles the batching internally. Multiple calls are unnecessary overhead.

7. Not considering the column's type when passing a Python int for a BIGINT.

Python ints are arbitrary precision. If you pass a number >2^31 to an INTEGER (4 bytes), an error. For a BIGINT (8 bytes), fine. Check the PostgreSQL types vs the Python ones.

8. Forgetting that COPY doesn't fire Python triggers.

If your model has @event.listens_for(Task, 'before_insert'), it does NOT run with COPY. PL/pgSQL triggers in the DB do. If you need pre-insert Python logic, COPY isn't the approach.

Exercise: implement the POST /bulk endpoint

Setup: a Task model with id, title, status, priority. The table created.

Step 1: implement the endpoint with copy_records_to_table.

@router.post("/tasks/bulk")
async def bulk_import(tasks: list[TaskCreate], db = Depends(get_db)):
    # ... implement it
    pass

Step 2: measure the performance with different N.

# Generates a payload of N tasks
async def benchmark(N):
    payload = [{"title": f"T{i}", "status": "pending", "priority": i % 5} for i in range(N)]
    response = await client.post("/tasks/bulk", json=payload)
    return response.json()["elapsed_ms"]

for N in [100, 1000, 10000, 50000]:
    ms = await benchmark(N)
    print(f"N={N}: {ms}ms")

Step 3: add validation with error handling.

# If a task has an empty title, handle it gracefully
class TaskCreate(BaseModel):
    title: str = Field(min_length=1)
    status: str
    priority: int = Field(ge=1, le=5)

Test with an invalid payload:

curl ... -d '[{"title": "", "status": "pending", "priority": 5}]'
# 422 Unprocessable Entity (Pydantic validation)

Step 4: add streaming for large cases.

If you receive >10k tasks, consider streaming from a generator instead of loading everything into a list.

async def stream_records(tasks: list[TaskCreate]):
    for t in tasks:
        yield (t.title, t.status, t.priority)


await conn.copy_records_to_table(..., records=stream_records(tasks))

Step 5: document the limits.

Set an explicit limit (max_items=50000) and return a clear error if it's exceeded.

N=100: 5ms
N=1000: 12ms
N=10000: 145ms
N=50000: 680ms

Summary and next step

What you learned:

• The 3 COPY variants in asyncpg: copy_records_to_table (memory), copy_to_table (a file), copy_from_query (export).
• copy_records_to_table is the default for in-memory data from Python.
• Access to the raw asyncpg connection from SQLAlchemy: (await session.connection()).get_raw_connection().driver_connection.
• Serialization: common types are automatic; datetime requires a timezone; arrays as lists; JSON as a dict.
• Error handling: COPY is atomic; validate first with Pydantic, or use a temporary table + a filter.
• Streaming for large datasets with generators.
• Typical performance: 10k rows in ~150ms, 50k in ~700ms.

Before moving on, you should be able to:

• Implement a POST /bulk endpoint with copy_records_to_table.
• Handle COPY errors with an explicit rollback.
• Decide between upfront validation (Pydantic) vs post-COPY validation (a temp table).
• Apply explicit limits for protection.

In the next capsule we go to the intermediate approach: bulk_insert_mappings and its limitations. When it beats COPY (when you need standard SQL parameterization, when the model is complex). When it loses because of its limitations (skipping events, Python defaults). And the typical confusion with the SQLAlchemy 2.0 syntax that changed from earlier versions.

Resources

1. asyncpg — copy_records_to_table — the official reference.
2. asyncpg — copy_to_table — for files.
3. PostgreSQL Docs — COPY — the command's reference.
4. SQLAlchemy 2.0 — Async raw connection access — the pattern for raw access.
5. Brandur Leach — Postgres COPY in production — a real case.
6. Heap — High performance writes — extreme scale.
7. Citus Data — COPY benchmarks — comparisons.

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