Async Batch Ingestion with psycopg3 Connection Pooling

Sustaining tens of thousands of telemetry rows per second into a hypertable is less about raw insert speed and more about balancing three coupled limits: how many concurrent connections the pool holds, how many rows each COPY moves per round trip, and how much unwritten data you let pile up in memory before applying backpressure. This page turns that balance into arithmetic — profile the target rate, size the pool and batch against COPY latency, and wire producers to workers through a bounded queue so the write path never outruns PostgreSQL. It builds on the batching and idempotency work in the parent guide on Python ETL orchestration.

# Input Profiling: Metrics to Gather First

Every sizing decision below flows from five measurable quantities. Capture them from a representative production window before touching pool or batch settings:

• Target ingest rate — sustained rows/sec the pipeline must absorb at peak, not the daily mean. IoT feeds are bursty; size to the busy hour.
• Row footprint — average bytes per row on the wire in binary COPY format (roughly the sum of column widths plus a few bytes of per-field framing). Governs memory and network cost per batch.
• Batch latency — wall-clock time for one worker to stream a batch of a given size with COPY and commit. Measure it against the real hypertable, because chunk routing, indexes, and triggers all add to it.
• Pool size — the number of concurrent connections the ingest process is allowed to hold open.
• Database max_connections — the hard ceiling shared across every client, background worker, and replication slot on the instance.

The pool must be a small fraction of max_connections. TimescaleDB reserves connections for the job scheduler, continuous aggregate refresh workers, compression jobs, and superuser reservation; an ingest pool that grabs the remainder starves them and blocks the very lifecycle automation the data depends on.

# Sizing the Pool and Batch Against COPY Latency

Sustained throughput is a product of how much each worker moves per commit and how many workers run in parallel:

$$T_{rows/s} = \frac{B_{batch}}{L_{batch}} \times N_{workers}$$

where $B_{batch}$ is rows per batch, $L_{batch}$ is the mean seconds to stream and commit one batch, and $N_{workers}$ is the number of concurrent COPY workers — which, since each worker holds exactly one connection while it copies, is also the required pool size. Rearranging for the pool size needed to hit a target:

$$N_{workers} = \left\lceil \frac{T_{target} \times L_{batch}}{B_{batch}} \right\rceil$$

The two knobs trade off. Larger $B_{batch}$ amortizes per-round-trip and per-transaction overhead, so throughput per connection climbs — but only until the batch grows large enough to bloat WAL and lengthen the lock window on each chunk. Larger $N_{workers}$ adds parallelism cheaply until it approaches the instance’s effective write concurrency (bounded by CPU, WAL flush, and disk), after which added connections just queue on locks and buffers. The goal is the smallest $N_{workers}$ that meets $T_{target}$, leaving connection headroom for everything else.

Cap the pool so it never crowds the server:

$$N_{workers} \le \left\lfloor \left(\text{max\_connections} - C_{reserved}\right) \times 0.5 \right\rfloor$$

with $C_{reserved}$ covering the scheduler, refresh and compression workers, replication, and superuser reservation (budget 15–25 on a typical instance). Holding the ingest pool at or below half the remaining connections leaves room for read traffic and ad-hoc sessions.

# The AsyncConnectionPool and a binary COPY worker

psycopg v3 ships the pool separately in the psycopg_pool package. Open it once, size min_size/max_size to $N_{workers}$, and let each worker borrow a connection only for the duration of its COPY. Binary COPY (with_types on a %b statement) skips text encoding and parsing, which is measurably faster for wide numeric telemetry:

import asyncio
from psycopg_pool import AsyncConnectionPool

POOL_SIZE = 8          # == N_workers, from the sizing formula
BATCH_SIZE = 10_000    # rows per COPY, from the latency trade-off
QUEUE_CAP = POOL_SIZE * 3   # bounded backlog -> backpressure

COPY_SQL = (
    "COPY sensor_readings (time, device_id, metric_type, value) "
    "FROM STDIN WITH (FORMAT BINARY)"
)


async def copy_batch(pool: AsyncConnectionPool, batch: list[tuple]) -> None:
    """Stream one batch into the hypertable with a binary COPY."""
    async with pool.connection() as conn:          # borrow for COPY only
        async with conn.cursor() as cur:
            async with cur.copy(COPY_SQL) as copy:
                copy.set_types(["timestamptz", "uuid", "text", "float8"])
                for record in batch:
                    await copy.write_row(record)
        # exiting the `async with pool.connection()` commits and returns it

Because the connection is released at the end of the async with block, one connection is held per in-flight batch and no longer — so pool size, worker count, and peak concurrency are the same number by construction.

# Backpressure with a bounded queue

An unbounded queue turns a slow database into an out-of-memory crash: producers keep appending faster than workers drain. A bounded asyncio.Queue fixes this — once it hits maxsize, await queue.put(...) suspends the producer until a worker frees a slot, so pressure propagates back to the source instead of into the heap:

async def worker(pool: AsyncConnectionPool, queue: asyncio.Queue) -> None:
    """Accumulate rows into a batch, then COPY it."""
    batch: list[tuple] = []
    while True:
        record = await queue.get()
        if record is None:                # sentinel: drain and stop
            if batch:
                await copy_batch(pool, batch)
            queue.task_done()
            return
        batch.append(record)
        if len(batch) >= BATCH_SIZE:
            await copy_batch(pool, batch)
            batch = []
        queue.task_done()


async def run(dsn: str, producer) -> None:
    queue: asyncio.Queue = asyncio.Queue(maxsize=QUEUE_CAP)
    async with AsyncConnectionPool(
        dsn, min_size=POOL_SIZE, max_size=POOL_SIZE, open=False
    ) as pool:
        await pool.wait()                 # fail fast if the DB is unreachable
        workers = [
            asyncio.create_task(worker(pool, queue)) for _ in range(POOL_SIZE)
        ]
        async for record in producer:     # blocks here when the queue is full
            await queue.put(record)
        for _ in workers:                 # one sentinel per worker
            await queue.put(None)
        await asyncio.gather(*workers)

Sizing the queue at roughly POOL_SIZE * BATCH_SIZE * 3 rows keeps enough buffered to feed every worker across a COPY cycle without letting the backlog grow unbounded. If producers routinely block on put, the database — not Python — is the bottleneck, which is exactly the signal you want.

# Worked Example

A platform must sustain 50,000 rows/sec of four-column telemetry into a hypertable. Benchmarking one worker against the real table shows a 10,000-row binary COPY commits in about 120 ms, so per-worker throughput is:

$$\frac{B_{batch}}{L_{batch}} = \frac{10{,}000}{0.12} \approx 83{,}300 \text{ rows/s per worker}$$

The required worker count, and therefore pool size, is:

$$N_{workers} = \left\lceil \frac{50{,}000 \times 0.12}{10{,}000} \right\rceil = \lceil 0.6 \rceil = 1$$

A single worker theoretically covers the target, but running at 100% of one connection’s capacity leaves no margin for latency spikes, chunk-boundary crossings, or a concurrent compression job. Provision for headroom instead: at 8 workers the pipeline has roughly 13× the headline capacity, absorbs bursts to ~660k rows/sec, and still fits comfortably under a modest connection budget:

$$N_{workers} = 8 \le \left\lfloor (100 - 20) \times 0.5 \right\rfloor = 40$$

With max_connections = 100 and ~20 reserved, an 8-connection ingest pool uses a fraction of the ceiling. The batch stays at 10,000 rows — large enough to amortize round trips, small enough that each transaction writes well under a chunk’s worth of WAL. Pair the ingest pipeline with the idempotent upsert patterns for late-arriving telemetry when duplicates from gateway retries are possible, since plain COPY cannot deduplicate on its own.

# Edge Cases & When to Deviate

The model assumes append-mostly writes into a well-sized hypertable. Adjust when:

• Pool starvation — if workers block waiting to borrow a connection, max_size is below the worker count, or the app leaks connections by holding them outside the COPY block. Keep min_size == max_size == N_workers and release inside the async with so borrow time is bounded.
• Oversized transactions and WAL pressure — very large batches inflate WAL, lengthen replication lag, and hold locks on the target chunk longer. If pg_wal growth or checkpoint frequency spikes, cut $B_{batch}$ and add a worker to hold throughput constant.
• COPY vs executemany crossover — COPY wins decisively above a few hundred rows per batch. For small trickle writes, or when you need ON CONFLICT handling that COPY lacks, a batched cursor.executemany() (or COPY into a staging table then an INSERT ... SELECT ... ON CONFLICT) is the better tool. Sizing the chunk the data lands in still matters here — see how to calculate the optimal chunk_interval for IoT sensor data.
• Connection lifetime — long-lived pooled connections accumulate cached plans and memory. Set max_lifetime on the pool (e.g. 30 minutes) so connections recycle and the backend’s memory footprint stays flat over multi-day runs.
• Writes spanning many chunks — back-loading historical data hits many chunks at once, multiplying index maintenance and lock churn. Sort each batch by time so a COPY targets as few chunks as possible, and prefer a larger chunk_time_interval during bulk backfill.

# Verification

Confirm the pipeline sustains its target and that connection use stays within budget. Sample ingest rate straight from the hypertable:

-- Rows landed in the last minute -> observed rows/sec.
SELECT count(*) / 60.0 AS rows_per_sec
FROM sensor_readings
WHERE time > now() - INTERVAL '1 minute';

Check that the ingest pool is not crowding the connection ceiling, and watch for sessions stuck waiting on locks — the tell-tale sign of oversized transactions or too many workers:

-- Active connections vs the configured ceiling.
SELECT
    count(*)                                        AS total_conns,
    count(*) FILTER (WHERE state = 'active')        AS active,
    current_setting('max_connections')::int         AS max_conns,
    count(*) FILTER (WHERE wait_event_type = 'Lock') AS waiting_on_locks
FROM pg_stat_activity
WHERE datname = current_database();

If waiting_on_locks climbs during ingest, reduce batch size or worker count; if total_conns approaches max_conns, the pool is oversized relative to the rest of the workload. A healthy pipeline shows observed rows/sec at or above target, a stable connection count well under the ceiling, and near-zero lock waits.

# Related

• Python ETL Orchestration — the parent guide on building resilient TimescaleDB ingestion in Python
• Idempotent Upsert Patterns for Late-Arriving Telemetry — deduplicating retried and out-of-order writes that COPY cannot handle alone
• How to Calculate Optimal chunk_interval for IoT Sensor Data — sizing the chunks your batches land in
• Chunk Compression Scheduling Automation — compressing the chunks once ingestion fills them
• TTL Policy Mapping & Enforcement — expiring ingested data on a predictable schedule

← Back to Python ETL Orchestration

Up to Data Retention, Compression & Lifecycle Automation