Python ETL Orchestration

The engineering problem this guide solves is the unglamorous glue between a fleet of IoT gateways and a TimescaleDB hypertable: how a Python worker reliably lands telemetry at rate, corrects late or duplicated readings without doubling rows, and backfills a reconnecting gateway’s buffer — all without exhausting connections or fighting the compression and retention jobs running underneath it. The database primitives are well documented; the loader that feeds them usually is not. This page assembles that loader with psycopg v3 and its pool, and wires it into the data retention and compression lifecycle so ingestion and maintenance cooperate rather than collide.

A hypertable will happily accept one INSERT per reading, and that naive path is exactly what falls over first: a per-row round trip caps throughput at a few thousand rows per second, opens and closes a connection per burst, and offers no defence against the duplicate deliveries and out-of-order arrivals that define real telemetry. The loader below replaces that path with pooled connections, COPY-based batches, conflict-aware corrections, and a bounded backfill that never sprays unbounded ranges at the refresh scheduler.

# Prerequisites

The loader assumes psycopg v3 with its companion pool package, a hypertable that already exists, and enough background-worker and connection headroom that ingestion does not starve the maintenance jobs. Confirm each item before wiring the pipeline.

• psycopg[binary]>=3.1 and psycopg_pool>=3.2 installed in a Python 3.11+ environment; the pool ships separately from the driver.
• A target hypertable exists with a right-sized chunk_time_interval — over-fragmented chunks turn every batch into a many-chunk write and inflate lock churn.
• Any unique constraint you intend to use for upserts includes the partitioning column (time); a hypertable cannot enforce a unique index that omits its partition key.
• max_connections on the server leaves room for the pool’s max_size on top of the background workers, replication, and admin sessions.
• timescaledb.max_background_workers and max_worker_processes are high enough that chunk compression scheduling and refresh jobs still run while the loader is saturated.
• statement_timeout is set (session or role level) so a stuck COPY cannot pin a pooled connection indefinitely.
• The connecting role has INSERT on the hypertable and USAGE on its schema; the backfill step additionally needs rights to call refresh_continuous_aggregate.

Verify the driver, the pool package, and the server’s connection ceiling in one pass:

import psycopg, psycopg_pool
print("psycopg", psycopg.__version__, "pool", psycopg_pool.__version__)

with psycopg.connect("postgresql://etl@db/telemetry") as conn:
    with conn.cursor() as cur:
        cur.execute("SHOW max_connections;")
        print("max_connections", cur.fetchone()[0])
        cur.execute("SHOW timescaledb.max_background_workers;")
        print("bg_workers", cur.fetchone()[0])

# Step-by-Step Implementation

The four steps map to the diagram: you stand up the pool that fronts every write, drain the queue into the hypertable with COPY, switch to INSERT ... ON CONFLICT for the correction path, then close the loop with a bounded backfill that refreshes aggregates over exactly the range it touched.

# 1. Build a connection pool

Opening a fresh connection per batch dominates the cost of ingestion and, under a reconnect storm, will exhaust max_connections outright. A pool amortises the handshake across the worker’s lifetime and caps concurrency at a value the server can actually serve. Create it once at process start and keep it open — the pool is the batch-loader node in the diagram.

from psycopg_pool import ConnectionPool
from psycopg.rows import dict_row

# One pool per worker process, opened at startup and closed at shutdown.
pool = ConnectionPool(
    conninfo="postgresql://etl@db/telemetry",
    min_size=4,            # warm connections kept ready for bursts
    max_size=16,           # hard ceiling; must fit under server max_connections
    max_idle=300,          # seconds before an idle-above-min connection is trimmed
    timeout=10.0,          # raise PoolTimeout rather than block forever
    kwargs={"row_factory": dict_row, "autocommit": False},
    open=True,
)

For an event loop rather than a thread pool, swap in AsyncConnectionPool from the same package; its async with pool.connection() as conn API mirrors the synchronous one and is the right choice when the queue consumer is already asyncio-based. Either way, borrow a connection only for the duration of a batch and return it promptly — the async batch ingestion and pooling patterns go deeper on sizing min_size/max_size against gateway concurrency.

# 2. Batch ingest via COPY

COPY is the throughput path. It streams rows over the wire in a single logical statement, bypassing the per-statement parse/plan overhead that caps executemany, and on a hypertable it routes each row to its chunk exactly as a plain insert would. psycopg v3 exposes it through cursor.copy() as a context manager you feed row tuples with write_row:

from datetime import datetime

def ingest_batch(rows: list[tuple[datetime, str, str, float]]) -> int:
    """Land a batch of (time, device_id, metric_type, value) tuples via COPY.
    Returns the number of rows written."""
    with pool.connection() as conn:            # borrowed from the pool, returned on exit
        with conn.cursor() as cur:
            with cur.copy(
                "COPY sensor_readings (time, device_id, metric_type, value) "
                "FROM STDIN"
            ) as copy:
                for row in rows:
                    copy.write_row(row)        # type-adapted client-side, streamed to server
        conn.commit()                          # one commit per batch, not per row
    return len(rows)

write_row adapts each Python value to its PostgreSQL type using the driver’s binary protocol, so a datetime lands as timestamptz and a float as double precision without string formatting. Keep batches in the low thousands of rows: large enough to amortise the round trip, small enough that a failure re-sends little and the transaction holds few locks. This is the step-2 edge into the hypertable in the diagram.

# 3. Idempotent upsert via INSERT … ON CONFLICT

COPY has no conflict handling — a duplicate delivery becomes a duplicate row. For the correction path (a gateway resending a window it already sent, or a sensor revising a reading) use INSERT ... ON CONFLICT, which requires a unique constraint. On a hypertable that constraint must include the time partitioning column; a unique index on (device_id, metric_type) alone is rejected. Define it once:

-- The unique constraint MUST contain the partition key (time) on a hypertable.
ALTER TABLE sensor_readings
  ADD CONSTRAINT sensor_readings_uniq UNIQUE (device_id, metric_type, time);

Then upsert with executemany, which psycopg v3 pipelines efficiently for parameterised statements. ON CONFLICT ... DO UPDATE makes a replayed batch converge to one row per key instead of accumulating duplicates:

def upsert_batch(rows: list[tuple]) -> None:
    """Idempotent correction path: a replayed reading overwrites rather than duplicates."""
    with pool.connection() as conn:
        with conn.cursor() as cur:
            cur.executemany(
                """
                INSERT INTO sensor_readings (time, device_id, metric_type, value)
                VALUES (%s, %s, %s, %s)
                ON CONFLICT (device_id, metric_type, time)
                DO UPDATE SET value = EXCLUDED.value
                WHERE sensor_readings.value IS DISTINCT FROM EXCLUDED.value;
                """,
                rows,
            )
        conn.commit()

The WHERE ... IS DISTINCT FROM guard skips a heap write when the incoming value matches what is stored, so idempotent replays stay cheap. Route bulk first-delivery traffic through COPY (step 2) and reserve this path for retries and corrections — the idempotent upsert patterns for late-arriving telemetry cover conflict targets, partial indexes, and update-vs-nothing trade-offs in detail.

# 4. Bounded backfill that triggers a refresh

When a gateway reconnects after an outage it dumps a buffer spanning hours. Replaying it in one transaction holds locks too long and, worse, silently leaves any continuous aggregate stale over that span. Process the buffer in bounded time slices, track the min and max timestamp actually written, and refresh aggregates over only that closed range — the dashed arc in the diagram:

from datetime import datetime, timedelta

def backfill_and_refresh(rows: list[tuple], cagg: str = "sensor_readings_hourly",
                         slice_hours: int = 6) -> None:
    """Replay a reconnect buffer in bounded slices, then refresh the aggregate
    over exactly the min..max range that was written."""
    if not rows:
        return
    rows.sort(key=lambda r: r[0])                 # order by time
    lo, hi = rows[0][0], rows[-1][0]

    # 1. Land the buffer in bounded chunks so no single txn spans the whole range.
    for start in _slices(lo, hi, timedelta(hours=slice_hours)):
        end = start + timedelta(hours=slice_hours)
        window = [r for r in rows if start <= r[0] < end]
        if window:
            upsert_batch(window)                  # idempotent: safe to retry a slice

    # 2. Reconcile the aggregate over the affected, closed range only.
    with pool.connection() as conn:
        conn.autocommit = True                    # refresh cannot run in a txn block
        with conn.cursor() as cur:
            cur.execute(
                "CALL refresh_continuous_aggregate(%s, %s, %s);",
                (cagg, lo, hi + timedelta(hours=1)),   # end is exclusive; cover the last bucket
            )

def _slices(lo: datetime, hi: datetime, step: timedelta):
    cur = lo
    while cur <= hi:
        yield cur
        cur += step

Bounding the refresh to lo..hi keeps the reconciliation cost proportional to the backfill, not to the whole hypertable, and dovetails with the refresh policy design and scheduling that handles steady-state materialisation. The gateway-reconnect backfill automation extends this into watermark tracking and resumable slices.

# Configuration Parameters Reference

Size max_size from arrival rate and latency rather than by guessing. By Little’s law, the average number of connections in concurrent use is arrival rate $\lambda$ (batches per second) times mean service time $W$ (seconds per batch, including commit):

$$C_{\text{active}} = \lambda \cdot W, \qquad \text{max\_size} \ge \lceil \lambda \cdot W \rceil + C_{\text{headroom}}$$

For 30 batches/second at a 120 ms mean service time, $C_{\text{active}} = 30 \times 0.12 = 3.6$ connections in flight; a max_size of 6–8 leaves headroom for jitter without over-subscribing the server. Provisioning far above this only trades queueing in the pool for contention in the database.

# Integration with Adjacent Features

The loader is one stage of a lifecycle, and its settings are constrained by the stages on either side.

• Compression. Batches that write only recent timestamps land in write-hot chunks and never touch the columnar store, which is exactly what you want: chunk compression scheduling should seal a chunk only after ingestion for its range has stopped. Keep compress_after comfortably larger than your worst-case backfill horizon so a reconnect never has to write into an already-compressed chunk.
• Continuous aggregates. The step-4 refresh reconciles a continuous aggregate only over the backfilled range; steady-state buckets are still driven by their own policy. Bounding the manual refresh avoids the full-range rematerialisation that stalls the scheduler.
• Out-of-order writes. Telemetry that arrives late — the norm for cellular gateways — is a hypertable feature, not an error, but it interacts with chunk sizing and compression. Review handling out-of-order data insertion so the loader’s late writes route to the right chunk rather than forcing a decompress.

# Performance Validation

After the loader has run under load, confirm it is actually landing rows at rate and that the pool is not the bottleneck. Measure ingestion throughput straight from the hypertable over a recent window:

-- Rows landed per second over the last 10 minutes.
SELECT
    count(*)                                            AS rows_10m,
    round(count(*) / 600.0, 1)                          AS rows_per_sec,
    max(time)                                           AS newest_reading,
    now() - max(time)                                   AS ingest_lag
FROM sensor_readings
WHERE time > now() - INTERVAL '10 minutes';

A steady ingest_lag near zero means the loader keeps pace; a growing lag means batches are queueing. Cross-check the pool itself — if borrowers wait, max_size is too small or service time too long:

# psycopg_pool exposes live counters via get_stats().
stats = pool.get_stats()
print("in use:", stats["pool_size"] - stats["pool_available"],
      "waiting:", stats["requests_waiting"],
      "avg wait ms:", round(stats.get("requests_wait_ms", 0) / max(stats.get("requests_num", 1), 1), 1))

A non-zero requests_waiting under normal load is the signature of pool saturation: borrowers are blocked on timeout, and either max_size must rise (if the server has slots) or batches must get faster. Confirm the server has room before raising it — SELECT count(*) FROM pg_stat_activity WHERE usename = 'etl'; against max_connections.

# Troubleshooting

psycopg_pool.PoolTimeout: couldn't get a connection after N sec. Every pooled connection is checked out longer than timeout. Either the pool is undersized for the arrival rate (raise max_size up to the server’s spare max_connections) or a batch is stuck — add a statement_timeout so a wedged COPY releases its connection instead of pinning it. Check pool.get_stats()["requests_waiting"] to distinguish saturation from a leak.

COPY fails with invalid input syntax or column ... is of type timestamptz but expression is of type text. A row tuple’s Python types do not match the column types. write_row adapts by type, so pass a real datetime (not an ISO string) for time and a numeric for value; a stray None in a NOT NULL column raises here too. Validate and coerce tuples before the COPY loop rather than mid-stream, since an error aborts the whole batch.

ON CONFLICT raises there is no unique or exclusion constraint matching the ON CONFLICT specification. The conflict target does not match an existing unique constraint — most often because the constraint omits the time partitioning column, which a hypertable requires. Recreate it to include time (see step 3) and make the ON CONFLICT (...) column list match it exactly.

ERROR: cannot update/delete rows from chunk ... compression is enabled during an upsert. A correction is targeting an already-compressed chunk. This means compress_after is shorter than your correction horizon. Widen it so corrections land before compression, or decompress the specific chunk, apply the write, and let the policy recompress — never disable compression table-wide to force the write through.

Throughput plateaus far below expectation despite spare CPU. The loader is committing per row or per tiny batch. Confirm one commit() per batch, batch sizes in the thousands, and COPY (not executemany) on the first-delivery path. A pool max_size of 1 also serialises everything — verify the pool opened with the intended size via pool.get_stats().

Frequently Asked Questions

# Related & Navigation

← Back to Data Retention, Compression & Lifecycle Automation

• Chunk Compression Scheduling & Automation — the compression stage the loader must not write into prematurely.
• TTL Policy Mapping & Enforcement — the retention stage that drops the chunks this pipeline fills.
• Async Batch Ingestion with psycopg3 Connection Pooling — sizing and draining the pool under concurrent gateways.
• Idempotent Upsert Patterns for Late-Arriving Telemetry — conflict targets and update-vs-nothing semantics for corrections.
• Automating Backfill for IoT Gateway Reconnects — watermark tracking and resumable slices for reconnect buffers.