PyReverseETL

The open operational data activation runtime: move data from a source system into the tools your team actually works in, with a real lineage graph and real compliance enforcement.

What it is

PyReverseETL is 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 and not silently faked.

Features

• Real connectors, real I/O. PostgreSQL, MySQL, S3/S3-compatible object storage (MinIO) as source and destination; generic-schema read/write, incremental reads, upsert.
• Real HTTP destinations. Webhook destination with real auth (Bearer / API key / Basic) and JSON payload construction; CRM/marketing-automation REST adapters for Salesforce, HubSpot, Marketo, and Squid-CRM.
• Connector plugin SDK (Python path). A documented pyreverseetl.connectors entry-point group plus real local SQLite and optional DuckDB source/destination connectors, and an optional Postgres psycopg skeleton (run_with_connector). Third-party plugins are Python-level; the Rust run_sync engine still owns the built-in Postgres/MySQL/S3/webhook/CRM path.
• Compliance/PII handling before data leaves the process. Mask, remove, or truncate fields per-record; Encrypt is honestly reported as unresolved rather than faked.
• Lineage tracking. Every sync registers real source/destination nodes and appends a real edge (record count + timestamps) to an in-process lineage graph, queryable and exportable to JSON or Graphviz DOT.
• Reliability in the real sync path. Retry with backoff on transient HTTP failures, dry-run preview, schema-drift detection, and an idempotency ledger for crash-mid-batch recovery.
• CLI + Python API + optional Flask REST server, all backed by the same engine.

Who it's for

• Data and platform engineers syncing operational data (Postgres/MySQL) into a CRM or marketing tool, with compliance/PII masking applied before it leaves the source.
• Teams pushing to a webhook destination with real auth for integrations that don't have a dedicated connector yet.
• Anyone who needs to audit what moved where — every sync records a real lineage edge you can query afterward, not just a log line.

What it is not

• Not a streaming/CDC platform. Kafka, CDC streaming, and HDFS are not wired into the real sync path and are out of scope for this pass.
• Not a Spark/PySpark transform engine. The transformer is present from earlier work but not wired into execute/run_sync; several such modules return fixed, non-real numbers if called directly (documented inline, e.g. SparkTransformer::submit).
• Python connector plugins are not loaded by the Rust core. run_sync accepts only its fixed built-in connector set. Plugins registered under pyreverseetl.connectors run through the Python-level run_with_connector path, which does not write the Rust lineage graph. See Connector plugins.
• StatGuardian gates are CLI-level, not core-level. execute-activation supports a live --statguardian-url sample gate and a file-based --validation-report gate. The Rust run_sync core itself does not call StatGuardian. A real cross-repo test lives in tests/test_statguardian_live.py (skipped when StatGuardian is not installed).
• Not verified against live CRM accounts. The Salesforce/HubSpot/Marketo/Squid adapters are real REST clients, but they were verified against local mock HTTP servers that assert the exact request shape (no live account was available). Please report request-shape mismatches against a real account as issues. The Squid standard-object path is create-only; it does not update an existing person/company record.
• Not encryption. Encrypt compliance actions are reported as unresolved rather than silently claimed to succeed.

Connector status

Connector	Direction	Status
PostgreSQL	source + destination	Real, via sqlx. Generic-schema read/write, incremental reads, upsert. Verified with a real Postgres container.
MySQL	source + destination	Real, via sqlx. Same capabilities as Postgres. Verified with a real MySQL container.
S3 / S3-compatible object storage (MinIO, etc.)	source + destination	Real, via aws-sdk-s3, with a custom-endpoint / path-style option for MinIO. JSON-lines and CSV implemented; Parquet/Avro/ORC/Iceberg/Delta return an explicit "not implemented" error rather than silently no-op'ing. Verified with a real MinIO container.
Webhook	destination	Real HTTP POST/PATCH/DELETE via reqwest, with real auth headers (Bearer/API key/Basic) and JSON payload construction.
SQLite (python/pyreverseetl/connectors.py)	source + destination	Real local reads/writes via stdlib sqlite3, exposed through the Python connector-plugin path (run_with_connector), not the Rust run_sync engine.
DuckDB (python/pyreverseetl/connectors.py)	source + destination	Optional real local reads/writes via duckdb, including append/replace/upsert, tested against temporary database files. Python plugin path only; install with PyReverseETL[duckdb].
Postgres (python/pyreverseetl/connectors.py)	source + destination	Optional psycopg skeleton for a local server, including read/append/replace/upsert. Python plugin path only; not live-warehouse support. Install with PyReverseETL[postgres].
CRM (core/src/adapters/salesforce.rs)	destination	Real REST API client: OAuth2 token exchange, object-based create/upsert-by-external-ID/delete/describe. Verified against a local mock HTTP server.
CRM (core/src/adapters/hubspot.rs)	destination	Real v3-style REST API client (create, upsert-by-ID-property, delete, properties/schema). Verified against a mock server.
Marketing automation (core/src/adapters/marketo.rs)	destination	Real REST API client (identity token endpoint, bulk lead create/update, delete, describe). Verified against a mock server.
CRM (core/src/adapters/squid.rs)	destination	Real Squid-CRM REST create for people/companies with a stable per-record Idempotency-Key and X-Correlation-Id; response replay, body-hash conflict handling, Bearer auth, and allowlisted payloads. Verified against a mock server. Standard-object creates do not update existing records.
Other cloud object storage backends	source + destination	Not implemented. Returns an explicit error instead of a fake success.
StatGuardian gate	pre-sync gate	Wired in the CLI execute-activation path (--statguardian-url live /validate, or --validation-report); not in the Rust run_sync core.
Kafka, HDFS, Spark/PySpark transforms, CLI dashboard	—	Present from earlier work, out of scope for this pass, not wired into execute/run_sync. Treat as experimental.

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 adapter targets is named in the source file, not repeated here to avoid implying vendor endorsement.

Project status

Beta. Version 3.5.3. The Postgres/MySQL/S3/webhook paths are real and verified against real services; the web source connector (MudFish-backed site crawling) is implemented and unit-tested but not yet exercised end-to-end against a live site; the CRM/marketing adapters are real but mock-verified only; several modules from earlier work remain unwired and are called out as such above. For a blunt, file-and-line-referenced list of what's tested vs. untested vs. broken vs. not built, see docs/internal/ROADMAP_HONEST.md. Changes are tracked in CHANGELOG.md.

Installation

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. Requires Python 3.9+.

Quick start

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":"****"}}]'

Singer target protocol

Consume Singer NDJSON and forward RECORD batches to a sink:

cat singer-output.ndjson | pyreverseetl-singer-target --output records.jsonl

from pyreverseetl.singer import run_target, jsonl_sink
run_target(lines, jsonl_sink("records.jsonl"))

Supported messages: SCHEMA, RECORD (batched) and STATE (flushes the batch). This is a protocol adapter; native Rust connectors remain the execution path.

Connector plugins (Python path)

The Rust engine's connector set is fixed, so third-party connectors use a separate Python-level interface. A plugin subclasses BaseConnector and implements read(config, limit) and write(config, rows):

from pyreverseetl.connectors import (
    BaseConnector, get_connector, register_connector, run_with_connector,
)

class MyConnector(BaseConnector):
    name = "my_connector"

    def read(self, config, limit=None):
        return [{"id": 1, "name": "Alice"}]  # real I/O here

    def write(self, config, rows):
        return len(rows)  # real I/O here; raise instead of faking a count

register_connector("my_connector", MyConnector)
result = run_with_connector(
    "sqlite",
    {"path": "source.db", "table": "customers"},
    "sqlite",
    {"path": "destination.db", "table": "customers"},
    compliance_rules=[
        {
            "id": "mask_email",
            "rule_type": "pii_masking",
            "target_fields": ["email"],
            "action": {"type": "mask", "pattern": "****"},
        }
    ],
)
print(result.rows_read, result.rows_written, result.compliance_violations)

run_with_connector applies the same mask/remove/truncate actions as the core compliance engine; encrypt is reported as an unresolved violation, never faked. Set dry_run=True to read and apply compliance rules without calling the destination's write.

Package a plugin by declaring an entry point (no core changes needed):

[project.entry-points."pyreverseetl.connectors"]
my_connector = "my_package.connectors:MyConnector"

The bundled sqlite connector is a real source/destination backed by stdlib sqlite3 (append, replace, and upsert with upsert_key). The optional duckdb connector mirrors that behavior and is tested against temporary DuckDB databases:

pip install "PyReverseETL[duckdb]"

from pyreverseetl.connectors import run_with_connector

result = run_with_connector(
    "duckdb",
    {"path": "source.duckdb", "table": "customers"},
    "duckdb",
    {"path": "destination.duckdb", "table": "customers", "upsert_key": "id"},
)
print(result.rows_read, result.rows_written)

The optional postgres connector is a psycopg skeleton for a local server; configure it with a dsn/database_url or host/database/user/password fields:

pip install "PyReverseETL[postgres]"

result = run_with_connector(
    "postgres",
    {"dsn": "postgresql://user:pass@localhost:5432/demo", "table": "customers"},
    "postgres",
    {"dsn": "postgresql://user:pass@localhost:5432/demo", "table": "customers_copy"},
)

Neither Python driver is bundled. Missing drivers raise a clear install hint, and Postgres tests skip unless PYREVERSEETL_TEST_POSTGRES_DSN points at a reachable local server. Kafka, HDFS, Spark/PySpark transforms, and the CLI dashboard remain experimental/unwired; no Python plugin path changes that.

Changelog

3.5.0

• Optional DuckDBConnector (real local reads/writes) and PostgresConnector (local server via psycopg) for the Python connector plugin path, registered in the built-in registry and exported from pyreverseetl.

3.4.0

• pyreverseetl.connectors plugin interface with the pyreverseetl.connectors entry-point group.
• Real local SQLite source/destination connector (stdlib sqlite3) wired into the Python-level run_with_connector path, including compliance actions and dry-run previews.
• Tests use temporary SQLite databases for real reads/writes; the Rust run_sync path is unchanged.

3.3.1

• Encrypted CLI state (Fernet/keyring) with legacy plaintext migration.
• Key export/import for backup and recovery.
• Singer SCHEMA/RECORD/STATE target adapter.

Encrypted local state

Workflow/activation credentials are no longer stored as plaintext JSON. The CLI state file is encrypted with Fernet; the key lives in the OS keyring, or in PYREVERSEETL_STATE_KEY for headless CI:

export PYREVERSEETL_STATE_KEY="$(python -c 'from cryptography.fernet import Fernet;print(Fernet.generate_key().decode())')"

Existing plaintext state is read and transparently re-encrypted on the next save. Losing the key means losing the local state file (re-enter credentials); the error is explicit rather than silently falling back to plaintext.

Usage

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

Squid-CRM activation

The squid destination posts a real create to /rest/people or /rest/companies with a stable per-record Idempotency-Key. Reusing the same source record and payload after a transient failure replays Squid's stored response instead of creating a duplicate; reusing the key with a different body returns a non-retried conflict.

import json
import pyreverseetl

result = pyreverseetl.run_sync(
    source_type="postgres",
    source_config=json.dumps({
        "host": "localhost", "port": 5432, "database": "warehouse",
        "username": "postgres", "password": "postgres", "table": "customers",
    }),
    destination_type="squid",
    destination_config=json.dumps({
        "base_url": "http://localhost:3000",
        "object": "people",
        "idempotency_key_prefix": "warehouse-people",
        "auth": {"type": "bearer", "token": "squid-api-key"},
    }),
)
print(result.rows_written, result.rows_failed)

The field allowlist is intentionally small: people forward first_name, last_name, email, phone, job_title, and company_id; companies forward name and domain. Other source columns are not sent. Standard Squid object creates are create-only, so a changed source record needs an explicit update workflow rather than being silently re-created.

Reliability features

• Retry with backoff: every HubSpot/Salesforce/Marketo/Squid/webhook HTTP call retries transient failures (connection errors, timeouts, 429 rate limits) with exponential backoff. Squid retries reuse the same Idempotency-Key.
• 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.

  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).

  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) while still sending records whose content genuinely changed.

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

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.

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) and server.py (an optional Flask REST wrapper around the same engine).

Validation gate

# File-based report
pyreverseetl execute-activation my-activation --validation-report statguardian-report.json

# Live StatGuardian service: validate a sample before the real sync
pyreverseetl execute-activation my-activation --statguardian-url http://127.0.0.1:8000

A file report must have "passed": true; anything else blocks the sync. The live --statguardian-url gate dry-runs a sample and calls the service's /validate; a failure or network error blocks the real sync. It is a sample gate, not a full-table scan.

Documentation

• docs/internal/ROADMAP_HONEST.md — what's tested vs. untested vs. broken vs. not built, plus known technical debt.
• docs/ROADMAP.md — narrative roadmap.
• CHANGELOG.md — release history.
• COMPLIANCE_CHECKLIST.md — compliance posture.
• SECURITY.md — how to report a vulnerability.

Known issues and deferred work

Blocked / failed:

• Live Salesforce/HubSpot/Marketo/Squid verification (no accounts).
• Scheduler/run-history observability.
• CDC/streaming.

Rejected / deferred:

• Spark transforms and streaming CDC now — outside the proven sync path.

Full shipped/blocked/rejected detail: docs/internal/QUEUE_STATUS.md.

Contributing

See CONTRIBUTING.md for guidelines and CODE_OF_CONDUCT.md for community standards.

# Rust: build, test, lint
cargo build -p pyreverseetl-core
cargo test -p pyreverseetl-core --lib          # unit tests (hermetic, no external services)
cargo clippy -p pyreverseetl-core --lib
cargo fmt

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

The Docker-backed connector round-trip tests 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

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

Related projects

Other tools from the same family, for adjacent parts of the data platform:

• StatGuardian — fast, declarative data quality framework (Rust): schema validation, drift detection, anomaly detection.
• PyDBTGuard — analyzes a dbt project's manifest.json to flag risky tests, map blast radius, and estimate test cost from static metadata.
• PyAirflowTester — dependency intelligence and quality assurance for Airflow and dbt pipelines.
• PyBeamGuard — Apache Beam and Dataflow pipeline analysis for pre-deployment validation, cost forecasting, hot-key detection, and reliability.
• ClusterAudienceKit — Rust-powered customer segmentation: RFM, clustering, churn, CLV, and lookalike audiences.
• PyWeatherEnriched — real historical weather and geocoding enrichment for ML pipelines.

Browse all projects →

License

This project is licensed under the Apache License 2.0.