PyReplicatorX

Lightweight, open-source Python library for syncing data from databases to cloud warehouses. Think of it as a Python-native alternative to Airbyte — no JVM, no heavy infrastructure, just pip install and go.

Features

• CDC (Change Data Capture) — Stream real-time changes from PostgreSQL (WAL2JSON) and MySQL (binlog)
• Full & Partial Dumps — One-shot table dumps with optional date-range filtering
• Cloud Warehouse Destinations — Load into Redshift, Snowflake, or BigQuery
• Parquet Staging — Efficient columnar format staged on S3 or GCS before loading
• Schema Evolution — Auto-detect new columns and evolve destination schemas
• Checkpointing — Resume from where you left off after restarts (SQLite-based)

Architecture

Source (CDC/Dump) → Queue → Processor → Stager (Parquet → S3/GCS) → Destination (COPY INTO)
     │                                       │                            │
     └── Checkpoint Store (SQLite)           └── Schema Manager ──────────┘

All sources emit a common NormalizedRecord format. The pipeline batches records per table, writes Parquet files with Snappy compression, uploads to cloud storage, then issues native COPY/Load commands to the destination warehouse.

Supported Connectors

Quick Start

Install

# Core (PostgreSQL source + Redshift destination + S3 staging)
pip install pyreplicatorx

# With MySQL support
pip install pyreplicatorx[mysql]

# With Snowflake destination
pip install pyreplicatorx[snowflake]

# With BigQuery destination
pip install pyreplicatorx[bigquery]

# Everything
pip install pyreplicatorx[all]

Configure

Create a config file (e.g., config.json):

{
  "pipeline_name": "my_sync",
  "source": {
    "engine": "postgres",
    "host": "localhost",
    "port": 5432,
    "user": "postgres",
    "password": "MY_DB_PASSWORD_ENV_VAR",
    "database": "mydb",
    "sslmode": "prefer",
    "slot_name": "pyreplicatorx_slot",
    "tables": ["public.users", "public.orders"]
  },
  "destination": {
    "engine": "redshift",
    "host": "my-cluster.region.redshift.amazonaws.com",
    "port": 5439,
    "user": "admin",
    "password": "MY_REDSHIFT_PASSWORD_ENV_VAR",
    "database": "analytics",
    "schema": "public",
    "iam_role": "arn:aws:iam::123456789:role/RedshiftCopyRole"
  },
  "staging": {
    "type": "s3",
    "bucket": "my-staging-bucket",
    "prefix": "pyreplicatorx/staging",
    "region": "us-east-1"
  },
  "settings": {
    "batch_size": 1000,
    "batch_timeout_seconds": 30
  }
}

Passwords reference environment variable names (e.g., MY_DB_PASSWORD_ENV_VAR) that are resolved at runtime from .env files or the shell environment.

Run

# Stream CDC changes (long-running)
pyreplicatorx cdc -f config.json

# Dump tables (one-shot)
pyreplicatorx dump -f config.json

# Validate config without running
pyreplicatorx validate -f config.json

Dump Configuration

Full and partial dumps are configured in the source.dump_tables array:

{
  "source": {
    "dump_tables": [
      "public.users",
      {
        "table": "public.events",
        "date_column": "created_at",
        "start_date": "2025-01-01",
        "end_date": "2025-12-31"
      }
    ]
  }
}

• String entry — Full table dump
• Object entry — Partial dump filtered by date column

Development

# Clone and install in editable mode
git clone https://github.com/your-org/pyreplicatorx.git
cd pyreplicatorx
pip install -e ".[dev]"

# Start test PostgreSQL (port 5433)
docker compose -f tests/test_dbs/postgres/docker-compose.yml up -d

# Format
black --line-length 200 pyreplicatorx/

# Lint
ruff check pyreplicatorx/

# Test
pytest

License

MIT