PyReverseETL

Reverse ETL: move data from a source system into the tools your team actually works in, with a real lineage graph and real compliance enforcement.

A Rust sync engine with Python bindings and a CLI. pyreverseetl execute opens a real connection to a real source, reads real records, runs them through a real compliance/PII-masking engine, writes them to a real destination, and records a real lineage edge (source, destination, record count, timestamps) that you can query afterward. There is no simulated mode: if a connector isn't wired to a real backend yet, it's not offered as an option, not silently faked.

What's real right now

If you need a connector marked "not implemented" above, that's an honest gap, not a documentation oversight — open an issue rather than assuming it works. The exact third-party API each CRM/marketing-automation adapter targets is named in the source file, not repeated here to avoid implying vendor endorsement.

Reliability

All four wired into the real production sync path (execute_sync), not a separate/dead one:

• Retry with backoff: every HubSpot/Salesforce/Marketo/webhook HTTP call (upsert, delete, schema fetch, OAuth token exchange) automatically retries transient failures (connection errors, timeouts, 429 rate limits) with exponential backoff, instead of failing on the first blip.
• Dry run (--dry-run / dry_run=True): reads from the real source and runs the real compliance engine, but never writes to the destination. dry_run_preview shows the exact payload each record would have sent, so you can audit before committing.

  pyreverseetl execute my_activation --dry-run
  

• Schema-drift detection (--schema-store <path>): persists the last-seen field-name/type shape for a source->destination pair to a real SQLite file, and reports any field added/removed/type-changed since the last run (schema_changes) — instead of only finding out via a per-record HTTP error from the destination.

  pyreverseetl execute my_activation --schema-store .pyreverseetl/schema.db
  

• Idempotency ledger (--idempotency-store <path>): persists which exact record content was already sent to which destination to a real SQLite file. Re-running the same sync (e.g. after a crash mid-batch) skips records already synced (rows_skipped_idempotent) instead of re-sending them, while a record whose content genuinely changed is still sent — this covers destinations with no upsert semantics of their own (the webhook adapter just POSTs) as well as crash-mid-batch recovery.

  pyreverseetl execute my_activation --idempotency-store .pyreverseetl/idempotency.db
  

Install

pip install pyreverseetl
# or
uv pip install pyreverseetl

pyreverseetl --version    # prints the installed version, read straight from the compiled Rust core

This installs a compiled Rust extension (built with maturin/PyO3) plus the pyreverseetl console command.

Quick start: a real sync against a local Postgres

This spins up a real Postgres container, seeds a table, and syncs it to a webhook using the real engine end to end.

docker run --rm -d -p 5432:5432 \
  -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=demo \
  --name pyreverseetl-demo-pg postgres:16

docker exec -i pyreverseetl-demo-pg psql -U postgres -d demo -c "
  CREATE TABLE customers (id INT PRIMARY KEY, name TEXT, email TEXT, ltv NUMERIC);
  INSERT INTO customers VALUES
    (1, 'Alice', 'alice@example.com', 4200.50),
    (2, 'Bob',   'bob@example.com',   1800.00);
"

# In another terminal: a throwaway HTTP endpoint to receive the synced rows
python3 -m http.server 8000 &   # or use https://webhook.site for a real inspectable URL

pyreverseetl create-workflow ltv_sync "LTV to webhook" postgres customers \
  --source-config '{"host":"localhost","port":5432,"database":"demo","username":"postgres","password":"postgres"}'

pyreverseetl create-activation ltv_to_hook ltv_sync webhook \
  --dest-config '{"url":"http://localhost:8000/hook","auth":{"type":"bearer","token":"demo"}}'

pyreverseetl execute ltv_to_hook
# {"status": "success", "run_id": "...", "rows_synced": 2, "rows_read": 2,
#  "rows_failed": 0, "compliance_violations": [], "duration_ms": 12,
#  "message": "Activation executed: 2 rows written to webhook"}

pyreverseetl lineage
# {"status": "success", "format": "json",
#  "lineage": {"nodes": {...}, "edges": [{"run_id": "...", "record_count": 2, ...}]}}

Mask PII before it ever leaves the process:

pyreverseetl execute ltv_to_hook --compliance-rules \
  '[{"id":"mask_email","rule_type":"pii_masking","target_fields":["email"],"action":{"type":"mask","pattern":"****"}}]'

Every one of these calls goes through the real Rust engine (pyreverseetl._core.run_sync) — rows_synced is the number of records the destination connector actually wrote, not a placeholder.

Python API

import json
import pyreverseetl

result = pyreverseetl.run_sync(
    source_type="postgres",
    source_config=json.dumps({
        "host": "localhost", "port": 5432, "database": "demo",
        "username": "postgres", "password": "postgres", "table": "customers",
    }),
    destination_type="webhook",
    destination_config=json.dumps({
        "url": "http://localhost:8000/hook",
        "auth": {"type": "bearer", "token": "demo"},
    }),
    limit=None,
    compliance_rules=None,
)
print(result.rows_written, result.duration_ms)

# Real lineage graph accumulated across every run_sync call in this process
print(pyreverseetl.lineage_json())
print(pyreverseetl.lineage_dot())  # Graphviz DOT export

Lineage tracking

Every run_sync call registers real source/destination nodes and appends a real edge — actual record count, actual start/completion timestamps — to an in-process lineage graph (pyreverseetl_core::lineage::LineageGraph). It supports upstream/downstream queries and exports to JSON or Graphviz DOT. This did not exist anywhere in the codebase before this pass; the README previously described "lineage tracking" as a feature with zero backing code.

Compliance & PII handling

DefaultComplianceEngine (pyreverseetl_core::governance::compliance_rules) applies real per-record rules before a write: mask a field, remove it, truncate it, or (for Encrypt) honestly report it as unresolved — there is no real encryption implementation, and the engine says so in check_compliance rather than silently claiming success. A MockComplianceEngine still exists but is #[cfg(test)]-only, so it can never run in a real build; it exists purely as a test double for exercising governance wiring without needing real masking behavior.

Architecture

Python CLI / API  →  pyreverseetl._core (PyO3 bindings)  →  pyreverseetl_core::execute_sync
                                                                  │
                                            ┌─────────────────────┼─────────────────────┐
                                       source read          compliance apply        destination write
                                  (postgres/mysql/s3)      (DefaultComplianceEngine)  (postgres/mysql/s3/
                                                                                       webhook/CRM/marketing-
                                                                                       automation adapters)
                                                                  │
                                                          lineage edge recorded

• core/ — the Rust engine: connectors, the compliance engine, lineage tracking, the sync executor.
• python/src/ — PyO3 bindings (run_sync, lineage_json, lineage_dot, plus the lower-level data-model classes PyWorkflow/PyDestination/etc.).
• python/pyreverseetl/ — the installed Python package: cli.py (the pyreverseetl command), server.py (an optional Flask REST wrapper around the same engine), both backed by the real engine rather than any in-process simulation.

Development

# Rust: build, test, lint
cargo build -p pyreverseetl-core
cargo test -p pyreverseetl-core --lib          # unit tests (hermetic, no external services)
cargo test -p pyreverseetl-core --lib -- --ignored   # real Docker-backed round-trip tests, see below
cargo clippy -p pyreverseetl-core --lib
cargo fmt

# Python bindings
maturin develop --release
pytest tests/ -v

Running the real, Docker-backed connector tests

Unit tests are hermetic by design (no network, no containers). The round-trip tests that prove the Postgres/MySQL/S3 connectors actually talk to a real service are #[ignore]d by default; run them explicitly against real containers:

docker run --rm -d -p 5439:5432 -e POSTGRES_PASSWORD=postgres \
    -e POSTGRES_DB=pyreverseetl_test --name pyreverseetl-pg-test postgres:16
docker run --rm -d -p 3307:3306 -e MYSQL_ROOT_PASSWORD=mysql \
    -e MYSQL_DATABASE=pyreverseetl_test --name pyreverseetl-mysql-test mysql:8
docker run --rm -d -p 9000:9000 -e MINIO_ROOT_USER=minioadmin \
    -e MINIO_ROOT_PASSWORD=minioadmin --name pyreverseetl-minio-test minio/minio server /data
docker run --rm --entrypoint sh minio/mc -c \
    "mc alias set local http://host.docker.internal:9000 minioadmin minioadmin && mc mb local/pyreverseetl-test"

PYREVERSEETL_TEST_PG_PORT=5439 PYREVERSEETL_TEST_MYSQL_PORT=3307 \
PYREVERSEETL_TEST_MINIO_ENDPOINT=http://localhost:9000 \
    cargo test -p pyreverseetl-core --lib -- --ignored

The same Docker services also back real, end-to-end Python-level tests in tests/test_real_sync_docker.py (they run the actual pyreverseetl CLI command as a subprocess and check real rows moved through it); those are skipped, not failed, when the containers aren't running.

Contributing

See CONTRIBUTING.md for guidelines.

Known gaps (deliberately out of scope for this pass)

• Other cloud object storage backends: return an explicit "not implemented" error.
• Kafka / CDC streaming, HDFS, PySpark transforms, the CLI dashboard, StatGuardian quality-gate integration: present in the codebase but not wired into the real sync path (execute / run_sync); several of these return fixed, non-real numbers if you call their APIs directly (this is documented inline in the affected modules, e.g. SparkTransformer::submit). Treat anything not listed in the connector table above as unverified.
• The CRM/marketing-automation adapters (core/src/adapters/) are real API implementations but were only verified against mocked HTTP responses (no live account was available in this environment) — please report any request-shape mismatches against a real account as issues.

License

Proprietary License. See LICENSE for details. All rights reserved.

Support

• GitHub Issues: PyReverseETL/issues
• Discussions: PyReverseETL/discussions