Metadata-Version: 2.2
Name: pyreplicatorx
Version: 0.1.0
Summary: Lightweight Python data sync library - CDC and dump replication to cloud warehouses
Author: Ammar Ibrahim
License: MIT
Project-URL: Homepage, https://github.com/ammar-bay/PyReplicatorX
Project-URL: Repository, https://github.com/ammar-bay/PyReplicatorX
Project-URL: Bug Tracker, https://github.com/ammar-bay/PyReplicatorX/issues
Keywords: cdc,data-sync,etl,replication,postgres,mysql,redshift,snowflake,bigquery
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: click>=8.0
Requires-Dist: psycopg2-binary>=2.9
Requires-Dist: pyarrow>=14.0
Requires-Dist: boto3>=1.28
Requires-Dist: python-dateutil>=2.8
Requires-Dist: prettyconf>=2.2
Provides-Extra: mysql
Requires-Dist: pymysqlreplication>=1.0; extra == "mysql"
Requires-Dist: pymysql>=1.0; extra == "mysql"
Provides-Extra: snowflake
Requires-Dist: snowflake-connector-python>=3.0; extra == "snowflake"
Provides-Extra: bigquery
Requires-Dist: google-cloud-bigquery>=3.0; extra == "bigquery"
Requires-Dist: google-cloud-storage>=2.0; extra == "bigquery"
Provides-Extra: all
Requires-Dist: pyreplicatorx[bigquery,mysql,snowflake]; extra == "all"
Provides-Extra: dev
Requires-Dist: pytest>=7.0; extra == "dev"
Requires-Dist: pytest-cov; extra == "dev"
Requires-Dist: black; extra == "dev"
Requires-Dist: ruff; extra == "dev"

# 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

| Source     | CDC | Full Dump | Partial Dump |
|------------|-----|-----------|--------------|
| PostgreSQL | Yes | Yes       | Yes          |
| MySQL      | Yes | Yes       | Yes          |

| Destination | Load Method                          | Schema Evolution       |
|-------------|--------------------------------------|------------------------|
| Redshift    | `COPY FROM S3 FORMAT AS PARQUET`     | `ALTER TABLE ADD COLUMN`|
| Snowflake   | `COPY INTO` with `MATCH_BY_COLUMN_NAME` | Native (auto)       |
| BigQuery    | GCS Load Job                         | `ALLOW_FIELD_ADDITION`  |

## Quick Start

### Install

```bash
# 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`):

```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

```bash
# 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:

```json
{
  "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

```bash
# 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
