• 9 min read
  • 1 code sample

asyncpg vs psycopg3 for HTS schedule bulk upserts

Choosing a PostgreSQL driver for HTS schedule loads is a throughput decision with a compliance tail: a full HTSUS revision is roughly 20,000 statistical-suffix rows, and each quarterly or mid-year release must land as an atomic, deduplicated upsert before any entry references it. This page benchmarks the two drivers customs ETL teams actually reach for — asyncpg and psycopg3 — across the three bulk-load paths that matter (server-side COPY, executemany, and execute_values-style batched inserts), and it gives you a runnable harness plus a decision matrix so you can pick a driver on measured throughput and memory rather than folklore. The short version: asyncpg’s copy_records_to_table wins raw ingest speed for staging, psycopg3’s Copy plus pipeline mode wins ergonomics and a cleaner INSERT ... ON CONFLICT path, and the right answer depends on whether your loader is already async and how much of the upsert logic you want in SQL versus Python. Both feed the bitemporal schema described in HTS Schedule Database Design.

Prerequisites

This comparison assumes a specific staging pattern and toolchain. Confirm each before running the harness:

  • PostgreSQL 14+ with a staging table and a target table. The canonical bulk-upsert pattern is COPY into an unlogged staging table, then a single INSERT ... SELECT ... ON CONFLICT DO UPDATE into the target — never per-row upserts across the wire. COPY cannot express ON CONFLICT, so staging is not optional for either driver.
  • Python 3.10+ with asyncpg>=0.29 and psycopg[binary,pool]>=3.1. The harness uses Decimal for general_rate and dataclass(frozen=True) rows; never use float for a duty rate, because ISO 4217 minor-unit rounding is not representable in binary floating point.
  • A normalized row shape. Keys are digit-only HTS codes (strip dots on write, exactly as the resolver in Handling missing HTS codes in ETL pipelines does on read) so lookups and conflict targets align. A mixed 8517.62.0000 / 8517620000 corpus silently splits into duplicate rows.
  • A parsed release in memory or on disk produced upstream by Tariff Update Ingestion Pipelines. This page loads an already-parsed release; it does not parse the WCO or USITC source.
  • Structured logging configured (stdlib logging with the %(asctime)s | %(levelname)s | %(name)s | %(message)s format) so per-path timing and row counts are greppable for audit.

Implementation

The harness below defines one immutable row type and one loader per path, then times each against the same batch. Every loader ends in the same idempotent INSERT ... ON CONFLICT DO UPDATE, so the only variable is how bytes reach the staging table. Read stage_asyncpg_copy and stage_psycopg_copy side by side — that pair is the whole benchmark.

import asyncio
import logging
import time
from dataclasses import dataclass
from datetime import date
from decimal import Decimal
from typing import Sequence

import asyncpg
import psycopg
from psycopg import sql

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("hts_bulk_upsert")

DDL = """
CREATE UNLOGGED TABLE IF NOT EXISTS hts_stage (
    hts_code       text        NOT NULL,
    description    text        NOT NULL,
    general_rate   numeric(8,4) NOT NULL,
    effective_date date        NOT NULL,
    expiry_date    date
);
CREATE TABLE IF NOT EXISTS hts_schedule (
    hts_code       text        NOT NULL,
    effective_date date        NOT NULL,
    description    text        NOT NULL,
    general_rate   numeric(8,4) NOT NULL,
    expiry_date    date,
    PRIMARY KEY (hts_code, effective_date)
);
"""

# Single idempotent merge, identical for both drivers. COPY cannot do ON CONFLICT,
# so every path stages first, then runs exactly this statement once.
MERGE = """
INSERT INTO hts_schedule (hts_code, effective_date, description, general_rate, expiry_date)
SELECT hts_code, effective_date, description, general_rate, expiry_date FROM hts_stage
ON CONFLICT (hts_code, effective_date) DO UPDATE
   SET description  = EXCLUDED.description,
       general_rate = EXCLUDED.general_rate,
       expiry_date  = EXCLUDED.expiry_date;
"""


@dataclass(frozen=True)
class HtsRow:
    hts_code: str            # digit-only, e.g. "8517620000"
    description: str
    general_rate: Decimal    # Decimal, never float
    effective_date: date
    expiry_date: date | None = None

    def as_tuple(self) -> tuple:
        return (self.hts_code, self.description, self.general_rate,
                self.effective_date, self.expiry_date)


def _records(rows: Sequence[HtsRow]) -> list[tuple]:
    # COPY column order for hts_stage.
    return [(r.hts_code, r.description, r.general_rate,
             r.effective_date, r.expiry_date) for r in rows]


# --- asyncpg: binary COPY straight from Python records ---------------------
async def stage_asyncpg_copy(dsn: str, rows: Sequence[HtsRow]) -> float:
    conn = await asyncpg.connect(dsn)
    try:
        await conn.execute("TRUNCATE hts_stage")
        t0 = time.perf_counter()
        # copy_records_to_table streams the binary protocol — no SQL text per row,
        # no client-side buffering of a giant VALUES string.
        await conn.copy_records_to_table(
            "hts_stage",
            records=_records(rows),
            columns=["hts_code", "description", "general_rate",
                     "effective_date", "expiry_date"],
        )
        await conn.execute(MERGE)
        elapsed = time.perf_counter() - t0
    finally:
        await conn.close()
    logger.info("asyncpg COPY+merge: %d rows in %.3fs", len(rows), elapsed)
    return elapsed


# --- asyncpg: executemany (prepared, per-row round trips) -------------------
async def stage_asyncpg_executemany(dsn: str, rows: Sequence[HtsRow]) -> float:
    conn = await asyncpg.connect(dsn)
    try:
        await conn.execute("TRUNCATE hts_stage")
        t0 = time.perf_counter()
        await conn.executemany(
            "INSERT INTO hts_stage VALUES ($1,$2,$3,$4,$5)", _records(rows)
        )
        await conn.execute(MERGE)
        elapsed = time.perf_counter() - t0
    finally:
        await conn.close()
    logger.info("asyncpg executemany+merge: %d rows in %.3fs", len(rows), elapsed)
    return elapsed


# --- psycopg3: Copy via the block protocol ---------------------------------
def stage_psycopg_copy(dsn: str, rows: Sequence[HtsRow]) -> float:
    with psycopg.connect(dsn) as conn, conn.cursor() as cur:
        cur.execute("TRUNCATE hts_stage")
        t0 = time.perf_counter()
        copy_sql = sql.SQL("COPY hts_stage ({}) FROM STDIN").format(
            sql.SQL(", ").join(map(sql.Identifier,
                ["hts_code", "description", "general_rate",
                 "effective_date", "expiry_date"]))
        )
        with cur.copy(copy_sql) as copy:
            for rec in _records(rows):
                copy.write_row(rec)   # streams row-by-row, constant memory
        cur.execute(MERGE)
        conn.commit()
        elapsed = time.perf_counter() - t0
    logger.info("psycopg3 COPY+merge: %d rows in %.3fs", len(rows), elapsed)
    return elapsed


# --- psycopg3: executemany with pipeline mode ------------------------------
def stage_psycopg_pipeline(dsn: str, rows: Sequence[HtsRow]) -> float:
    with psycopg.connect(dsn) as conn, conn.cursor() as cur:
        cur.execute("TRUNCATE hts_stage")
        t0 = time.perf_counter()
        # pipeline() removes the per-round-trip latency that cripples naive
        # executemany; still slower than COPY but far closer than without it.
        with conn.pipeline():
            cur.executemany(
                "INSERT INTO hts_stage VALUES (%s,%s,%s,%s,%s)", _records(rows)
            )
        cur.execute(MERGE)
        conn.commit()
        elapsed = time.perf_counter() - t0
    logger.info("psycopg3 pipeline+merge: %d rows in %.3fs", len(rows), elapsed)
    return elapsed


async def run_matrix(dsn: str, rows: Sequence[HtsRow]) -> None:
    await stage_asyncpg_copy(dsn, rows)
    await stage_asyncpg_executemany(dsn, rows)
    stage_psycopg_copy(dsn, rows)
    stage_psycopg_pipeline(dsn, rows)


if __name__ == "__main__":
    sample = [
        HtsRow(f"{85176200 + i:08d}00", f"Apparatus {i}",
               Decimal("0.0000"), date(2024, 1, 1))
        for i in range(20_000)
    ]
    asyncio.run(run_matrix("postgresql://localhost/customs", sample))

The load-bearing detail is that COPY never carries the upsert. Both copy_records_to_table (asyncpg) and cur.copy(...) (psycopg3) stream rows into unlogged staging at near-wire speed; the deduplication and rate-update logic lives entirely in the one MERGE statement, which runs once per release. That separation is what makes the load idempotent — re-running the same release is a no-op update, exactly the property Point-in-time HTS resolution with recursive CTEs relies on when it walks the effective-dated history.

Decision matrix: asyncpg versus psycopg3 for HTS bulk upserts A comparison matrix with six criteria rows and two driver columns, asyncpg and psycopg3. Fastest bulk path: asyncpg uses copy_records_to_table sending the binary protocol; psycopg3 uses cursor copy write_row streaming text. Raw COPY throughput: asyncpg is highest; psycopg3 is high and close behind. executemany without help: asyncpg is slow per-row, psycopg3 is slow unless pipeline mode is enabled. Memory profile: both stream at constant memory when using COPY row iterators. Concurrency model: asyncpg is async-only, psycopg3 offers both sync and async. Upsert ergonomics: both run INSERT ON CONFLICT after staging, psycopg3 has richer sql composition helpers. The bottom guidance row: choose asyncpg when the loader is already async and raw staging speed dominates; choose psycopg3 when you want one driver for sync and async code and cleaner SQL composition. asyncpg psycopg3 Fastest bulk path how rows reach staging copy_records_to_table binary protocol cur.copy() write_row streamed rows Raw COPY throughput highest high, close behind executemany (bare) slow, per-row slow → pipeline() fixes Memory (COPY path) constant / streamed constant / streamed Concurrency model async only sync AND async Upsert ergonomics ON CONFLICT in SQL + sql composition Pick asyncpg when the loader is already async and raw staging speed dominates; pick psycopg3 for one sync + async driver and richer SQL composition.
Six criteria against two drivers. Both stream at constant memory and both run the same ON CONFLICT merge; the split is asyncpg's raw COPY edge and async-only model versus psycopg3's dual sync/async model and SQL-composition helpers.

Verification steps

Run these checks against a representative release before standardizing on a driver:

  1. Warm-cache timing parity. Run run_matrix three times and discard the first (connection and prepared-statement warm-up). Compare the median COPY-path times; on a 20,000-row HTSUS release asyncpg’s copy_records_to_table should lead psycopg3’s cur.copy by a modest margin, and both should beat their respective executemany paths by an order of magnitude. A tie between COPY and executemany means the merge, not the load, is your bottleneck.
  2. Row-count identity. After each loader, assert SELECT count(*) FROM hts_schedule equals the distinct (hts_code, effective_date) count in the source. A shortfall means duplicate keys collapsed; an excess means dots were not stripped and 8517.62.0000 split from 8517620000.
  3. Idempotency. Run the same loader twice against the same release. The second run must change zero rows semantically — capture xmax or a RETURNING count and confirm the second pass is a pure update, never an insert. This is the property downstream point-in-time resolution depends on.
  4. Memory ceiling. Profile each loader with tracemalloc or /usr/bin/time -v. The COPY and pipeline paths must hold roughly constant resident memory as row count grows; if psycopg3’s executemany without pipeline() balloons, you left the pipeline context out.
  5. Decimal fidelity. Round-trip one row with general_rate = Decimal("0.0350") and assert the value read back is Decimal("0.0350"), not 0.035000000001. A drift here means a float was introduced between parse and load.
  6. Transaction atomicity. Kill the process between COPY and MERGE. The target table must be unchanged (staging is unlogged and discarded); confirm no half-loaded release is visible to readers.

Edge cases & gotchas

  • COPY cannot upsert — do not fake it. Neither driver’s COPY accepts ON CONFLICT. Teams sometimes pre-delete conflicting keys before COPY to dodge staging; that opens a window where an entry could resolve against a missing rate. Always stage-then-merge in one transaction.
  • asyncpg binds by position, psycopg3 by placeholder. asyncpg uses $1,$2; psycopg3 uses %s. Copy-pasting a query between the two silently breaks parameter binding, and a mis-bound numeric(8,4) can truncate a rate rather than error. Keep the SQL strings driver-specific.
  • Type adaptation differs for numeric. asyncpg maps PostgreSQL numeric to Decimal natively; psycopg3 does too, but only if you have not registered a custom loader that coerces to float. Verify the adapter, because a coerced rate corrupts duty math under 19 CFR reasonable care.
  • Naive executemany is a latency trap. asyncpg’s executemany and psycopg3’s without pipeline() pay one network round trip per row — fine for 50 rows, catastrophic for 20,000. If you cannot use COPY (for example, an ORM in the way), psycopg3’s pipeline() recovers most of the gap; asyncpg has no equivalent knob, so reach for COPY there.
  • Unlogged staging is not crash-durable — that is the point. An unlogged hts_stage is fast because it skips WAL, but it is truncated after an unclean shutdown. Never point a reader at staging; it exists only to feed the merge. The durable copy is the target table.
  • Connection pooling changes the picture at concurrency. These single-connection benchmarks measure the load path, not a live service. Under a pool (asyncpg.create_pool or psycopg_pool), psycopg3’s ability to serve sync and async callers from one pool can outweigh asyncpg’s per-load speed when the same service also reads schedules. Benchmark the workload you actually run, not just the bulk import.

Up: HTS Schedule Database Design