cross-cloud-data-sync

A Flask application plus Python library for moving data between cloud stores on a schedule or on demand. Source and target systems are plugins behind a single Connector interface; jobs are configured through a wizard in the UI or via YAML.

Problem domain

Most data teams accumulate a long tail of one-off scripts that copy data between systems — Postgres-to-Snowflake replication, S3-to-Azure DR mirroring, BigQuery-to-Oracle bridges for legacy reports, ad-hoc migration jobs. The patterns here consolidate that work into a single application a non-engineer can configure and operate.

Built-in connectors

• AWS S3 (parquet / csv / json)
• Azure Blob Storage
• GCP BigQuery
• Snowflake
• PostgreSQL
• Oracle

Each implements the same five methods (connect, close, read, write, schema_inspect, row_count). Adding a new system is a single file — see src/data_sync/plugins/postgres.py for the canonical example.

Layout

src/data_sync/
  config/        settings + logging
  models/        SQLAlchemy ORM (Job, JobRun, Reconciliation)
  core/          orchestrator, reconciler, db session
  plugins/       base.py (ABC + registry) + 5 connector modules
  web/           flask app, blueprints, jinja templates, static css
  cli.py         typer CLI (init-db, import-job, run-job, list-jobs)

Web UI

make run-web starts the server on :5000.

• / — dashboard with stat cards and recent runs
• /jobs — job list
• /jobs/new — three-step wizard (basics → source → target)
• /jobs/<id> — job detail and history
• /jobs/<id>/run — manual trigger with runtime params (watermark override, batch size, write mode)
• /runs/<id> — live run viewer; progress streamed over Server-Sent Events

CLI

data_sync init-db                              # creates state DB
data_sync import-job -f config/sample_*.yaml   # import a yaml-defined job
data_sync list-jobs
data_sync run-job -n postgres_to_snowflake_daily
data_sync recent-runs --limit 5

Job format (YAML)

name: postgres_to_snowflake_daily
source:
  type: postgres
  url: postgresql+psycopg2://user:pw@host:5432/db
  table: events
  cdc:
    type: timestamp
    column: updated_at
target:
  type: snowflake
  account: abc12345.us-east-1
  user: etl_user
  password: __PLACEHOLDER__
  database: ANALYTICS_DW
  schema: PUBLIC
  table: STG_EVENTS
transforms:
  - cast_columns:
      created_at: "datetime64[ns]"
batch_size: 10000
schedule: "0 2 * * *"

Running locally

cp .env.example .env
make install
make init-db
make run-web                # http://localhost:5000

Stack

Python 3.11, Flask, Jinja2, SQLAlchemy (state DB), APScheduler, pandas, polars, pyarrow, boto3, azure-storage-blob, google-cloud-bigquery, snowflake-connector-python, psycopg2, oracledb, pydantic, structlog.

Design notes

• The scheduler runs in-process. High-availability deployments should back APScheduler with a Postgres jobstore and run multiple gunicorn workers, or move scheduling out to Airflow.
• CDC for sources without a timestamp or version column is currently hash-based, which reads the full source table on each run. Adding native triggers (Postgres logical replication, Oracle CDC) is the intended next step.
• The default deployment has no RBAC layer. Production deployments typically place the UI behind an SSO proxy; a Flask-Login layer is straightforward to add for standalone use.
• Reconciliation currently compares row counts and schemas. Per-column distribution comparison is on the roadmap.

About this code

Open-source companion to the integration and migration work done by acilox. For paid implementation, custom connectors, or extension of this framework into a production environment, open an issue.