Postgres

pyrula.connectors.postgres is a thin Arrow IO surface over ADBC. It reads query results straight into a pyarrow.Table and ingests Arrow batches without a per-row Python hop. The pipeline adapters (PostgresSource, PostgresSink) build on it, and you can call it directly from a workflow step.

pip install 'pyrula-connectors[postgres]'

Like the Delta connector, the high-level calls return an Either so errors surface as values rather than exceptions. Each takes a DSN per call rather than holding a connection, and an Err carries a PostgresError.

Read

from pyrula.connectors.postgres import read_arrow

result = read_arrow(
    "postgresql://localhost/prod",
    "SELECT id, amount FROM orders WHERE region = $1",
    parameters=("us",),
)
table = result.value  # Ok(pyarrow.Table); check result.is_err() in production code

read_arrow(dsn, sql, parameters=()) runs the query and fetches the whole result as an Arrow table, returned as Ok(table) or Err(PostgresError). Use parameters for values; never interpolate them into the SQL.

Write

import pyarrow as pa
from pyrula.connectors.postgres import write_arrow

result = write_arrow(
    "postgresql://localhost/prod",
    "orders",
    pa.table({"id": [1, 2], "amount": [10, 20]}),
    mode="create_append",   # ADBC ingest mode
)
assert result.is_ok()

write_arrow(dsn, table, batch, *, mode="create_append") ingests an Arrow batch in one autocommit-off transaction and self-commits, returning Ok(None) or Err(PostgresError). It is a convenience for workflow steps and one-off ingests. Do not use it from an exactly-once sink: the PostgresSink keeps data and checkpoint in a single transaction it owns itself, so a self-committing write would split the two.

Table and column names are interpolated into SQL, so the connector validates them with validate_identifier(name) before use; an invalid identifier comes back as an Err.

Replay-safe upsert

write_arrow appends, so it is not safe to call from a durable workflow step: replay or recovery would re-run the step and double-write. For that, use upsert_arrow, the Postgres counterpart to DeltaTable.merge_keyed:

from pyrula.connectors.postgres import upsert_arrow

result = upsert_arrow(
    "postgresql://localhost/prod",
    "orders",
    batch,            # pyarrow Table or RecordBatch
    keys=["id"],      # conflict columns; a unique/PK constraint on them must exist
)
assert result.is_ok()

It stages the batch in a temp table and runs INSERT ... ON CONFLICT (keys) DO UPDATE in one transaction, updating every non-key column, and returns Ok(None) or Err(PostgresError). Re-running it with the same rows is a no-op, so a step that retries cannot double-write. Like write_arrow it self-commits, so do not call it from the exactly-once PostgresSink (which owns its own transaction).

Pipelines

The Pipelines connectors build on this. PostgresSink does the COPY and a checkpoint upsert in one Postgres transaction for exactly-once delivery; PostgresSource reads incrementally by a watermark column. See the postgres_to_delta recipe.