aurora-to-rs 0.1.5

Lean Aurora PostgreSQL to Redshift loader via copy_expert + S3 COPY

aurora_to_rs

Lean Aurora PostgreSQL to Redshift loader using psycopg2.copy_expert + S3 + Redshift COPY.

Bypasses pandas entirely -- ~10x faster than DataFrame-based approaches for large tables. IAM role auth only.

Install

pip install aurora_to_rs

Quick Start

from aurora_to_rs import aurora_to_rs

loader = aurora_to_rs(
    region_name='ap-south-1',
    s3_bucket='my-bucket',
    redshift_c=redshift_conn,          # psycopg2 connection (autocommit=True)
    postgres_engine=pg_engine,          # SQLAlchemy engine
    iam_role_arn='arn:aws:iam::123456:role/MY_ROLE'
)

3 Strategies

1. full_load -- Full table refresh (TRUNCATE + COPY)

For master/reference tables where filter_cond = '1=1'.

loader.full_load("SELECT * FROM master.items", "master.items")

2. delete_and_insert -- Delete by filter + COPY

For transaction tables with del_table = 'Y' (incremental by date/timestamp filter).

loader.delete_and_insert(
    "SELECT * FROM tran.shipment_au WHERE create_datetime >= current_date-3",
    "tran.shipment_au",
    "create_datetime >= current_date-3",
    min_timestamp='2026-02-19',          # optional safety overlap
    timestamp_col='create_datetime'
)

3. upsert -- Staging-based incremental upsert

For tables with key-based upsert (everything else). NULL-safe key matching.

loader.upsert(
    "SELECT * FROM tran.orders WHERE dt >= current_date-1",
    "tran.orders",
    ['order_id']
)

How It Works

Aurora (PG COPY TO STDOUT)  -->  BytesIO (strip \x00, parse header)  -->  S3  -->  Redshift COPY FROM

• Streams CSV via psycopg2.copy_expert -- no pandas DataFrame in the path
• Strips null bytes (\x00) from Aurora text columns automatically
• Cleans column names (/, ., - replaced with _) to match Redshift DDL
• S3 upload with retry (exponential backoff on SSL/connection errors)
• All methods return row_count for logging

API Reference

Method	Args	Returns	Use Case
full_load(sql, dest)	source SQL, dest table	row_count	Full refresh
delete_and_insert(sql, dest, filter, ...)	+ min_timestamp, timestamp_col	row_count	Incremental by date
upsert(sql, dest, keys)	+ upsert_columns list	row_count	Key-based upsert

Requirements

• Python >= 3.8
• boto3, psycopg2
• IAM role with S3 read/write and Redshift COPY permissions