pg-upsert¶

pg-upsert is a Python package for validating and upserting data from staging tables into base tables in PostgreSQL. It runs automated QA checks, reports errors with rich formatted output, and performs dependency-aware upserts.

An upsert (a blend of update and insert, standardized in SQL as MERGE) updates a row if it already exists and inserts it if it does not. pg-upsert matches staging rows to base rows by primary key — matched rows update the base table, unmatched rows are inserted — all inside a single transaction that is rolled back unless you ask to commit.

Why Use pg-upsert?¶

• 7 Automated QA Checks – Validates NOT NULL, PRIMARY KEY, UNIQUE, FOREIGN KEY, CHECK CONSTRAINT, column existence, and column type compatibility before any modifications occur.
• Interactive Confirmation – Two UI backends: Textual TUI (terminal) and Tkinter (desktop). Auto-detected or choose with --ui auto|textual|tkinter. The compare-tables dialog includes a Highlight Diffs toggle that tints matching/changed rows and flags the exact cells that differ, skipping any columns excluded from the upsert.
• Structured Results – run() returns an UpsertResult with per-table stats, QA errors, and JSON serialization (--output=json for CI/CD pipelines).
• Exportable Fix Sheets – --export-failures <dir> writes an actionable report of failing rows: one row per unique violating staging row with an _issues column listing every problem (NULL in 'genre', duplicate PK, FK violation, etc.) so users can open it in Excel and fix the data. Supports CSV (file per table), JSON (nested), and XLSX (sheets per table) via --export-format.
• Schema Validation – --check-schema flag validates column existence and type compatibility without running data checks or upserts.
• Flexible Upsert Strategies – Supports upsert, update, and insert methods.
• YAML Configuration – One config file drives both the CLI (--config-file) and the Python API (PgUpsert.from_config()), with layered sources and per-table column excludes.
• Dependency-Aware Ordering – Tables are processed in FK dependency order automatically.
• Rich Output – Colored pass/fail indicators, formatted tables, and dual console+logfile output.

Installation¶

Requires Python 3.10+ and a PostgreSQL database.

pip install pg-upsert

Optional extras:

pip install "pg-upsert[tui]"   # Textual terminal UI for interactive mode
pip install "pg-upsert[xlsx]"  # XLSX fix-sheet export

A prebuilt Docker image is also available — see Docker.

Usage¶

Python API¶

from pg_upsert import PgUpsert

result = PgUpsert(
    uri="postgresql://user@localhost:5432/mydb",
    tables=("genres", "publishers", "books", "authors", "book_authors"),
    staging_schema="staging",
    base_schema="public",
    do_commit=True,
    upsert_method="upsert",
    exclude_cols=("rev_user", "rev_time"),
    exclude_null_check_cols=("book_alias",),
    exclude_cols_by_table={"books": ("isbn_legacy",)},
    exclude_null_check_cols_by_table={"books": ("reprint_date",)},
).run()

# UpsertResult provides structured access to results
print(result.qa_passed)       # True if all QA checks passed
print(result.committed)       # True if changes were committed
print(result.total_updated)   # Total rows updated across all tables
print(result.total_inserted)  # Total rows inserted across all tables
print(result.to_json())       # JSON serialization for CI/CD

exclude_cols and exclude_null_check_cols apply to every table. The *_by_table mappings add per-table excludes merged on top of the global lists; each key must be one of the configured tables.

From a configuration file (the same YAML file the CLI accepts with --config-file):

from pg_upsert import PgUpsert

result = PgUpsert.from_config("pg-upsert.yaml").run()

# Keyword overrides beat file values, and multiple files layer left-to-right
result = PgUpsert.from_config(["base.yaml", "task.yaml"], do_commit=True).run()

Using an existing connection:

import psycopg
from pg_upsert import PgUpsert

conn = psycopg.connect(host="localhost", port=5432, dbname="mydb", user="user", password="pass")

ups = PgUpsert(
    conn=conn,
    tables=("genres", "publishers", "books"),
    staging_schema="staging",
    base_schema="public",
    do_commit=True,
)
result = ups.run()

# Drop all ups_* temp objects (connection stays open)
ups.cleanup()

QA-only mode (no upsert):

from pg_upsert import PgUpsert

ups = PgUpsert(
    uri="postgresql://user@localhost:5432/mydb",
    tables=("genres", "books"),
    staging_schema="staging",
    base_schema="public",
).qa_all()

# qa_passed is False only when ERROR-level findings exist.
# qa_errors returns ERROR findings only (block the upsert).
# qa_warnings returns WARNING findings only (informational, do not block).
# qa_findings returns all findings combined.
# qa_passed may be True while qa_warnings is non-empty.
if not ups.qa_passed:
    for err in ups.qa_errors:
        print(f"{err.table}: {err.check_type.value} — {err.details}")

Schema compatibility check (column existence and type mismatches only):

from pg_upsert import PgUpsert

ups = PgUpsert(
    uri="postgresql://user@localhost:5432/mydb",
    tables=("genres", "books"),
    staging_schema="staging",
    base_schema="public",
).qa_column_existence().qa_type_mismatch()

if ups.qa_errors:
    for err in ups.qa_errors:
        print(f"{err.table}: {err.check_type.value} — {err.details}")
else:
    print("Schemas are compatible")

Pipeline callbacks (per-table progress):

from pg_upsert import PgUpsert, CallbackEvent

def on_event(event):
    if event.event == CallbackEvent.QA_TABLE_COMPLETE:
        print(f"QA {'passed' if event.qa_passed else 'failed'} for {event.table}")
    elif event.event == CallbackEvent.UPSERT_TABLE_COMPLETE:
        print(f"{event.table}: {event.rows_inserted} inserted, {event.rows_updated} updated")

result = PgUpsert(
    uri="postgresql://user@localhost:5432/mydb",
    tables=("genres", "books"),
    staging_schema="staging",
    base_schema="public",
    do_commit=True,
    callback=on_event,
).run()

CLI¶

pg-upsert -h localhost -p 5432 -d mydb -u user \
  -s staging -b public \
  -t genres -t publishers -t books -t authors -t book_authors \
  -x rev_user -x rev_time \
  --commit

[!NOTE] CLI arguments take precedence over configuration file values. Explicit CLI flags are never overridden by the config file.

Configuration File¶

Create a YAML config file (see pg-upsert.example.yaml):

debug: false
commit: false
interactive: false
upsert_method: "upsert"  # Options: "upsert", "insert", "update"
logfile: "pg_upsert.log"
host: "localhost"
port: 5432
user: "docker"
database: "dev"
staging_schema: "staging"
base_schema: "public"
encoding: "utf-8"
tables:
  - "authors"
  - "publishers"
  - "books"
  - "book_authors"
  - "genres"
exclude_columns:
  - "rev_time"
  - "rev_user"
null_columns:
  - "book_alias"
exclude_columns_by_table:  # Per-table excludes, merged on top of exclude_columns
  books:
    - "isbn_legacy"
null_columns_by_table:  # Per-table NOT NULL check skips, merged on top of null_columns
  books:
    - "reprint_date"
output: "text"  # Options: "text", "json"
check_schema: false
compact: false
ui_mode: "auto"  # Options: "auto", "textual", "tkinter"
export_failures: null  # Directory to write QA failure fix sheet; null to disable
export_format: "csv"  # Fix sheet format: "csv", "json", or "xlsx"
export_max_rows: 1000  # Max rows captured per check per table for the fix sheet
strict_columns: false  # Treat all missing staging columns as errors

Run with: pg-upsert -f config.yaml, or load the same file in Python with PgUpsert.from_config("config.yaml").

Docker¶

docker run -it --rm \
  -v $(pwd):/app \
  ghcr.io/geocoug/pg-upsert:latest \
  -h host.docker.internal -p 5432 -d dev -u docker \
  -s staging -b public -t genres --commit

QA Checks¶

pg-upsert runs 7 types of QA checks on staging data before upserting:

[!NOTE] pg-upsert is constraint-driven. Data checks pass vacuously when the base table has no constraints of that type, and tables without a primary key are skipped during the upsert step (a warning is printed). To upsert against a table, make sure the base table has a PK. See Running Without Constraints for details.

See the QA Checks Reference for detailed documentation.

Authentication¶

pg-upsert resolves the database password in this order:

1. Password in URI (Python API only) — postgresql://user:pass@host/db
2. PGPASSWORD environment variable — standard PostgreSQL convention, works with both CLI and API
3. Interactive prompt — if neither of the above is set

For CI/CD pipelines, use PGPASSWORD to avoid interactive prompts:

PGPASSWORD=secret pg-upsert -h host -d db -u user \
  -s staging -b public -t books \
  --output json --commit

pg-upsert also supports PostgreSQL's .pgpass file via psycopg (libpq).

Exit Codes¶

Contributing¶

See CONTRIBUTING.md for development setup, available recipes, testing, and release process.

git clone https://github.com/geocoug/pg-upsert
cd pg-upsert
just sync
just test