ingestra 0.2.3

Universal SQL loader: extract from PostgreSQL, MySQL, BigQuery, or Redshift; align schema and upsert/overwrite into BigQuery, PostgreSQL, or MySQL—generated merge DDL, shared types.

Ingestra

Move data from one SQL system to another without hand-writing merge DDL. Ingestra is a small library for extract -> optional temp file step -> align schema -> upsert or overwrite across PostgreSQL, MySQL, BigQuery, and Redshift. The same DataSource + Ingest API works whether your warehouse is BigQuery or another relational database: one mental model, dialect-specific SQL generated for you.

Why teams adopt it fast

• One pipeline shape everywhere: stop rewriting ingestion logic for each engine pair.
• Schema drift handled for you: target create/alter follows source columns so migrations are less brittle.
• Merge behavior you control: choose upsert or overwrite, with lifecycle rules for target-only rows.
• Cloud and relational friendly: local NDJSON temp files for relational targets, cloud temp-load prep for BigQuery/Redshift.
• Practical for real ETL: dry_run for safe SQL previews, verbose for operational debugging.

The universal pattern

extract -> schema align -> delta prep -> upsert/overwrite

Keep this sequence, and you get predictable loads regardless of where data starts and where it lands.

---

Install

pip install .
pip install ".[postgres]"   # or mysql, bigquery, all

Run tests:

python3 -m unittest discover -s ingestra/tests -p "test_*.py" -v

---

Core Concepts

• DataSource: source or target implementation (columns, extract, extract_delta, optional target merge/schema methods).
• Ingest: orchestrates source + target and mirrors dry_run / verbose to both.
• *_delta: helper table used only to prepare one load before moving rows into the final table.
  • PostgreSQL/MySQL/Redshift: physical heap table.
  • BigQuery: external table over cloud storage files.
• LifecycleRule: controls target-only rows for overwrite (Delete, MarkAsDeleted, KeepUntouched).

Purpose of *_delta (very simple):

1. Put new rows in *_delta first.
2. Compare *_delta with the final table.
3. Insert/update/delete in the final table based on your merge mode.
4. Reuse *_delta for the next run.

It exists so loads are safer and easier to reason about than writing directly into the final table first.

---

Recommended Order of Operations

Local Targets (PostgreSQL, MySQL)

1. Extract (extract_to_local_file or extract_delta_to_local_file).
2. Align target schema (create_or_update_target_schema).
3. Load target delta from file (load_target_delta_from_ndjson_file).
4. Merge (upsert or overwrite).

Cloud Targets (BigQuery, Redshift)

1. Extract (to local file, or source-to-cloud when supported).
2. Align target main schema (create_or_update_target_schema).
3. Prepare target delta from cloud storage (prepare_target_delta_from_cloud_storage).
4. Merge (upsert or overwrite).

Quick routing

• If target is PostgreSQL/MySQL: use the local target flow.
• If target is BigQuery/Redshift: use the cloud target flow.
• In both cases, extraction completes first; merge-side work starts after that.

---

Flow A: PostgreSQL -> BigQuery

import datetime
import ingestra

source = ingestra.PostgreSQL("orders", schema_name="public")
source.connection(
    host="db.example.com",
    user="readonly",
    password="...",
    database="app",
    port=5432,
)

target = ingestra.BigQuery(
    "my-gcp-project",
    "staging",
    "orders",
    primary_keys=["order_id"],
)
target.connection("/path/to/service-account.json")

job = ingestra.Ingest(
    source,
    target,
    deletion_rule=ingestra.KeepUntouched(),  # or Delete(), MarkAsDeleted("is_deleted")
    dry_run=False,
    verbose=True,
)

# 1) Extract source delta to local NDJSON.
path = job.extract_delta_to_local_file(
    start_time=datetime.datetime.utcnow() - datetime.timedelta(days=3),
    end_time=datetime.datetime.utcnow(),
    column_name="updated_at",  # SQL sources require this for extract_delta
)

# 2) Upload NDJSON to GCS (outside Ingestra).

# 3) Align BigQuery main table schema.
job.create_or_update_target_schema()

# 4) Prepare BigQuery temporary loading table (`*_delta`) from GCS URIs.
job.prepare_target_delta_from_cloud_storage(
    ingestra.CloudStorageDeltaSpec(
        uris=["gs://bucket/orders/part-000.ndjson"],
    )
)

# 5) Merge staged delta into main table.
job.upsert()
# job.overwrite()

---

Flow B: MySQL <-> PostgreSQL (NDJSON staged)

Both relational targets use the same pattern: extract -> schema align -> load delta -> merge.

import datetime
import ingestra

source = ingestra.MySQL("orders", schema_name="app")
source.connection(host="legacy", user="ro", password="...", database="app")

target = ingestra.PostgreSQL(
    "orders",
    schema_name="warehouse",
    primary_keys=["order_id"],  # recommended on target
)
target.connection(
    host="pg.example.com",
    user="etl",
    password="...",
    database="analytics",
    port=5432,
)

job = ingestra.Ingest(source, target, deletion_rule=ingestra.Delete(), verbose=True)

# 1) Extract batch.
path = job.extract_delta_to_local_file(
    start_time=datetime.datetime.utcnow() - datetime.timedelta(days=1),
    end_time=datetime.datetime.utcnow(),
    column_name="updated_at",
)

# 2) Align target schema.
job.create_or_update_target_schema()

# 3) Load NDJSON into target temporary loading table (`*_delta`).
job.load_target_delta_from_ndjson_file(path)

# 4) Merge into main table.
job.overwrite()

Swap source/target classes for PostgreSQL -> MySQL; flow is unchanged.

---

Flow C: PostgreSQL -> Redshift (cloud staged)

import datetime
import ingestra

source = ingestra.PostgreSQL("orders", schema_name="public")
source.connection(host="db.example.com", user="readonly", password="...", database="app", port=5432)

target = ingestra.Redshift(
    "orders",
    schema_name="analytics",
    primary_keys=["order_id"],  # merge keys (no PK DDL assumptions)
)
target.connection(
    host="redshift.example.com",
    user="etl",
    password="...",
    database="warehouse",
    port=5439,
)

job = ingestra.Ingest(source, target, deletion_rule=ingestra.Delete(), verbose=True)

# 1) Extract source delta.
path = job.extract_delta_to_local_file(
    start_time=datetime.datetime.utcnow() - datetime.timedelta(days=1),
    end_time=datetime.datetime.utcnow(),
    column_name="updated_at",
)

# 2) Upload NDJSON to S3 (outside Ingestra).

# 3) Align Redshift main table schema.
job.create_or_update_target_schema()

# 4) Prepare Redshift temporary loading table (`*_delta`) via COPY from cloud storage.
job.prepare_target_delta_from_cloud_storage(
    ingestra.CloudStorageDeltaSpec(
        uris=["s3://your-bucket/orders/2026-04-20/"],
        credentials=ingestra.AwsIamRoleCredentials(
            iam_role="arn:aws:iam::123456789012:role/redshift-copy"
        ),
        region="eu-west-1",
        truncate=True,
    )
)

# 5) Merge into Redshift main table.
job.upsert()

Redshift can also export from source tables directly with:

• extract_to_cloud_storage(...)
• extract_delta_to_cloud_storage(...)

Both use UNLOAD and typed AWS credentials.

---

Flow D: Custom DataSource -> PostgreSQL (API-backed)

Use this when your source is not SQL-backed or have a custom integration.

Minimal shape:

import datetime
import ingestra

class MyApiSource(ingestra.DataSource):
    def columns(self, refresh=False) -> list:
        return [
            ingestra.Column("id", "VARCHAR", ""),
            ingestra.Column("value", "FLOAT", ""),
            ingestra.Column("updated_time", "TIMESTAMP", ""),
        ]

    def extract(self) -> list:
        return self.extract_delta(
            start_time=datetime.datetime.utcnow() - datetime.timedelta(days=365),
            end_time=datetime.datetime.utcnow(),
        )

    def extract_delta(self, start_time: datetime.datetime, end_time: datetime.datetime = None) -> list:
        # Fetch from API and yield normalized rows.
        yield self.extract_row({"id": "a", "value": 1.23, "updated_time": "2026-04-20 00:00:00"})

source = MyApiSource()
target = ingestra.PostgreSQL("metrics", schema_name="public", primary_keys=["id"])
# target.connection(...)

job = ingestra.Ingest(source, target, deletion_rule=ingestra.KeepUntouched(), verbose=True)

# 1) Extract source delta.
path = job.extract_delta_to_local_file(
    start_time=datetime.datetime.utcnow() - datetime.timedelta(days=7),
    end_time=datetime.datetime.utcnow(),
)

# 2) Align target schema.
job.create_or_update_target_schema()

# 3) Load NDJSON into target temporary loading table (`*_delta`).
job.load_target_delta_from_ndjson_file(path)

# 4) Merge staged delta into main table.
job.upsert()

Important: if your columns() list expands later, create_or_update_target_schema() propagates new columns to the target main table and target *_delta temporary loading table automatically.

---

Ingest Shortcuts

Method	Role	Use it for
extract_to_local_file(output_dir_name='/tmp')	Full extract -> NDJSON file path.	Any source -> temp-file workflow
extract_delta_to_local_file(start_time, end_time=None, *args, **kwargs)	Delta extract -> NDJSON file path; forwards source-specific delta args/kwargs.	Any source -> incremental temp-file workflow
create_or_update_target_schema(**kwargs)	Calls target create if empty, else alter.	All targets (PostgreSQL, MySQL, BigQuery, Redshift)
load_target_delta_from_ndjson_file(path, truncate=True, batch_size=500)	Loads NDJSON into target temporary loading table (*_delta).	Local targets (PostgreSQL, MySQL)
prepare_target_delta_from_cloud_storage(delta_spec)	Target-agnostic cloud temp-load prep.	Cloud targets (BigQuery, Redshift)
upsert(primary_keys=None)	Merge from temporary loading table (*_delta) without target-only delete behavior.	Targets with merge support (PostgreSQL, MySQL, BigQuery, Redshift)
overwrite(primary_keys=None)	Merge from temporary loading table (*_delta) and apply deletion_rule to target-only rows.	Targets with overwrite support (PostgreSQL, MySQL, BigQuery, Redshift)
extract_to_cloud_storage(...) / extract_delta_to_cloud_storage(...)	Source-side cloud export helpers.	Redshift source exports

---

Typed Configuration DSL + JSON

You can also define pipeline configuration as typed objects and load/save them as JSON.

import ingestra

cfg = ingestra.IngestraConfiguration(
    source=ingestra.MySQLConfiguration(
        table_name="table_name",
        schema_name="public",
        federated_connection="source_conn",
    ),
    target=ingestra.BigQueryConfiguration(
        project_name="project_name",
        dataset_name="dataset_name",
        table_name="table_name",
        primary_keys=["id"],
        federated_connection="target_conn",
    ),
    schedules=[
        ingestra.DeltaIngestion(
            frequency="hourly_3am_to_8pm",
            column_name="updated_at",
            lookback_seconds=7200,
            deletion_rule=ingestra.KeepUntouchedRuleConfiguration(),
        ),
        ingestra.SnapshotIngestion(
            frequency="never",
            deletion_rule=ingestra.DeleteRuleConfiguration(),
        ),
    ],
)

raw_json = cfg.to_json(indent=2)
cfg2 = ingestra.load_configuration_from_json(raw_json)

# Or from a JSON file path:
# cfg2 = ingestra.load_configuration_from_json_file("/path/to/config.json")

connections = {
    "source_conn": mysql_connection,
    "target_conn": bigquery_client,
}
ingest = cfg2.build(federated_connection=lambda name: connections[name])

Use the full-config loader above as the default path so app code has one JSON entrypoint.

JSON uses explicit typed objects (a type field) so it is deterministic to parse:

• source/target: mysql, postgresql, bigquery, redshift
• schedules: delta, snapshot
• lifecycle rules: keep_untouched, delete, mark_as_deleted

---

Capabilities by Engine

Area	PostgreSQL	MySQL	BigQuery	Redshift
Can export a whole table	[x]	[x]	[x]	[x]
Can export only new/changed rows	[x]	[x]	[x]	[x]
Can create/update the target table shape	[x]	[x]	[x]	[x]
Can load target table from cloud storage files	-	-	[x]	[x]
Can apply each load batch into the target table	[x]	[x]	[x]	[x]
Can export source data to cloud storage files	-	-	[ ]	[x]

Legend: [x] = supported, [ ] = not supported yet (but useful), - = not applicable.

---

Notes

• SQL-backed extract_delta requires a column_name (for time-window filtering).
• dry_run=True prints SQL and avoids execution for supported operations.
• verbose=True prints generated SQL during execution.

---

License

Copyright © 2019–2026 Datale Pte. Ltd.
Distributed under the MIT License.