pgsql_upserter

A powerful, production-ready PostgreSQL upsert utility with automatic schema introspection and intelligent conflict resolution. Perfect for serverless ETL pipelines and data integration workflows.

🚀 Key Features

• Zero Configuration: Automatic schema detection and column matching
• Intelligent Conflict Resolution: Automatically detects primary keys and unique constraints
• Production Tested: Handles deduplication, data validation, and error recovery
• Flexible Input: Supports both direct data (API responses) and CSV files

📦 Installation

pip install pgsql-upserter

Development Setup (uv)

uv sync --all-extras

🎯 Quick Start

Serverless ETL (Recommended)

Perfect for AWS Lambda, Google Cloud Functions, or any API-driven ETL:

from pgsql_upserter import UpsertEngine, create_connection_from_env

# Your API response data (Facebook Ads, Google Ads, etc.)
api_data = [
    {
        'account_id': '123456789',
        'campaign_id': 'camp_001', 
        'impressions': 1000,
        'clicks': 50,
        'spend': 25.50,
        'date_start': '2025-08-31'
    }
    # ... more records
]

# One function call does everything!
connection = create_connection_from_env()
result = UpsertEngine.upsert_data(
    connection=connection,
    data=api_data,  # Direct API data
    target_table='ads_metrics'
)

print(f"✅ {result.total_affected} rows processed")
print(f"📈 {result.rows_inserted} inserted, {result.rows_updated} updated")

CSV File Processing

# Automatic CSV processing
result = UpsertEngine.upsert_data(
    connection=connection,
    data='path/to/data.csv',  # File path
    target_table='ads_metrics'
)

🔧 Environment Setup

Set your PostgreSQL connection via environment variables:

export PGHOST=your-host
export PGPORT=5432
export PGDATABASE=your-db
export PGUSER=your-user
export PGPASSWORD=your-password

Or use a connection string:

export DATABASE_URL=postgresql://user:pass@host:port/dbname

🧠 How It Works

1. Schema Introspection: Analyzes your table structure automatically
2. Column Matching: Maps your data columns to table columns
3. Conflict Detection: Finds primary keys and unique constraints
4. Data Deduplication: Removes duplicates using conflict resolution strategy
5. Intelligent Upsert: Uses PostgreSQL's native INSERT...ON CONFLICT

🎯 Perfect For

• API Data Ingestion: Facebook Ads, Google Ads, LinkedIn Ads APIs
• Serverless ETL: AWS Lambda, Google Cloud Functions, Azure Functions
• Data Warehousing: Loading data into analytics databases
• Real-time Sync: Keeping databases in sync with external sources
• Batch Processing: Traditional CSV and file-based workflows

📊 Automatic Conflict Resolution

The library automatically chooses the best upsert strategy:

1. Primary Key: Uses table's primary key if available in data
2. Unique Constraints: Combines all unique constraints for conflict detection
3. Insert Only: Falls back to simple insert if no conflicts possible

🔍 Advanced Usage

Data Processing Before Upsert

from pgsql_upserter import UpsertEngine, read_csv_to_dict_list

# Read and process CSV data
csv_data = read_csv_to_dict_list('data.csv')

# Filter or transform data
filtered_data = [row for row in csv_data if float(row.get('spend', 0)) > 10.0]

# Upsert processed data
result = UpsertEngine.upsert_data(
    connection=connection,
    data=filtered_data,
    target_table='ads_metrics'
)

Custom Connection

import psycopg2
from pgsql_upserter import UpsertEngine

connection = psycopg2.connect(
    host="localhost",
    database="mydb",
    user="user",
    password="password"
)

result = UpsertEngine.upsert_data(
    connection=connection,
    data=your_data,
    target_table='your_table',
    schema='public'  # optional, defaults to 'public'
)

🛡️ Error Handling

The library provides comprehensive error handling and validation:

from pgsql_upserter import UpsertEngine, PgsqlUpserterError

try:
    result = UpsertEngine.upsert_data(connection, data, 'my_table')
    print(f"Success: {result.total_affected} rows processed")
except PgsqlUpserterError as e:
    print(f"Upsert failed: {e}")

✅ Testing & Coverage

Run tests with coverage:

uv run pytest

Current baseline:

• 41 tests passing
• 89.45% total coverage

📌 Project Status

• Current Version: 0.9.4
• Development Status: Pre-release/Beta
• Python Support: 3.11, 3.12, 3.13, 3.14
• Current Status: Actively maintained, no blocking issues, no known performance regressions

Recent highlights:

• Fixed PostgreSQL data type adaptation for dict and list values
• Added PostgreSQL type-aware JSON/array conversion in temp staging
• Standardized testing with pytest and measured coverage baseline
• Adopted uv workflow with hatchling build backend

🛠️ Tooling

• Dependency Management: uv
• Build Backend: hatchling
• Test Command: uv run pytest
• Coverage Artifacts: .coverage_html/, coverage.json

🗺️ Roadmap

1. Performance improvements: Optimize larger batch workloads and connection behavior
2. Advanced logging: Improve observability and operational diagnostics
3. Coverage hardening: Push targeted branch coverage beyond the current 89.45%

📋 Requirements

• Python 3.11-3.14
• PostgreSQL 12+
• psycopg2-binary

🤝 Contributing

Issues and pull requests are welcome! Please see our contributing guidelines.

Examples

Two runnable scripts are provided in the examples/ folder:

• examples/upsert_engine_demo.py — local execution demo showing both direct-data and CSV-path modes, fully driven by environment variables
• examples/serverless_etl_example.py — AWS Lambda/serverless ETL pattern using simulated Facebook Ads data, with proper JSON body serialization for API Gateway proxy responses

📚 See Also

• ARCHITECTURE.md for technical design and architectural decisions
• CHANGELOG.md for release history and notable changes

📄 License

MIT License - see LICENSE file for details.

🔗 Links

• PyPI Package
• Source Code
• Issues