pg-export-import

Stream-export filtered PostgreSQL rows to CSV and bulk-import via COPY. Supports multi-table, FK-aware pipelines with idempotent re-runs.

Installation

pip install pg-export-import

Requirements

• Python 3.11+
• PostgreSQL (source and target can be on different servers)

Library usage

Single table

from pg_export_import import ConnectionConfig, export_and_import

src = ConnectionConfig(host="db1.example.com", dbname="prod", user="app", password="...")
tgt = ConnectionConfig(host="db2.example.com", dbname="staging", user="app", password="...")

result = export_and_import(
    source_config=src,
    target_config=tgt,
    source_table="public.orders",
    target_table="public.orders",
    where_clause="created_at >= %s",
    where_params=("2024-01-01",),
    csv_path="/data/exports/orders.csv",   # optional; auto-generated if omitted
    delete_before_import=True,             # optional; delete target rows before importing (default False)
    # delete_where_clause="created_at >= %s",  # optional; defaults to where_clause
    # delete_where_params=("2024-01-01",),     # optional; defaults to where_params
)
print(result.status, result.exported_count, result.imported_count, result.deleted_count)

result.status is one of "success", "export_failed", "delete_failed", or "import_failed".

Multi-table pipeline

from pg_export_import import ConnectionConfig, run_pipeline

src = ConnectionConfig(...)
tgt = ConnectionConfig(...)

tables = [
    # parents before children (FK order)
    {"source_table": "customers",   "target_table": "customers",   "where_clause": ""},
    {"source_table": "orders",      "target_table": "orders",      "where_clause": "created_at >= %s", "where_params": ("2024-01-01",)},
    {"source_table": "order_items", "target_table": "order_items", "where_clause": "created_at >= %s", "where_params": ("2024-01-01",)},
]

results = run_pipeline(
    source_config=src,
    target_config=tgt,
    tables=tables,
    csv_dir="/data/exports",        # optional; defaults to tempdir
    fetch_size=5000,                # optional; rows per round-trip (default 5000)
    delete_before_import=True,      # optional; set False to append without deleting (default True)
)
for r in results:
    print(r["table"], r["status"], r["exported"], r["imported"])

CSV file naming

Each pipeline run captures a single timestamp once at start. All tables in that run share the same timestamp:

{csv_dir}/{source_table}_{YYYYMMDD_HHMMSS}.csv

Dots in schema-qualified names become underscores:

• The csv_dir is created automatically if it does not exist.
• On success, CSV files are deleted.
• On failure, CSV files are preserved for debugging or re-import.

Table config schema

Each entry in the tables argument to run_pipeline supports:

Security notes

• where_clause is embedded as a raw SQL fragment — it must come from trusted application code, never from user input.
• Runtime filter values must be passed via where_params, not interpolated into the clause string.
• All table and column identifiers are validated and safely quoted via psycopg.sql.Identifier.