Metadata-Version: 2.4
Name: ingestra
Version: 0.2.1
Summary: 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.
Author-email: "Datale Pte. Ltd." <hello@datale.io>
License: MIT
Project-URL: Homepage, https://github.com/dataleio/ingestra
Project-URL: Source, https://github.com/dataleio/ingestra
Keywords: bigquery,postgresql,mysql,redshift,etl,elt,ingestion,loader,merge,upsert,gcs,data-pipeline
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: Jinja2>=3.1
Provides-Extra: postgres
Requires-Dist: psycopg2-binary==2.9.5; extra == "postgres"
Provides-Extra: mysql
Requires-Dist: mysqlclient==2.1.1; extra == "mysql"
Provides-Extra: bigquery
Requires-Dist: cachetools==5.3.0; extra == "bigquery"
Requires-Dist: certifi==2022.12.7; extra == "bigquery"
Requires-Dist: charset-normalizer==3.1.0; extra == "bigquery"
Requires-Dist: google-api-core==2.11.0; extra == "bigquery"
Requires-Dist: google-auth==2.17.1; extra == "bigquery"
Requires-Dist: google-cloud-bigquery==3.9.0; extra == "bigquery"
Requires-Dist: google-cloud-core==2.3.2; extra == "bigquery"
Requires-Dist: google-crc32c==1.5.0; extra == "bigquery"
Requires-Dist: google-resumable-media==2.4.1; extra == "bigquery"
Requires-Dist: googleapis-common-protos==1.59.0; extra == "bigquery"
Requires-Dist: grpcio==1.53.0; extra == "bigquery"
Requires-Dist: grpcio-status==1.53.0; extra == "bigquery"
Requires-Dist: idna==3.4; extra == "bigquery"
Requires-Dist: packaging==23.0; extra == "bigquery"
Requires-Dist: proto-plus==1.22.2; extra == "bigquery"
Requires-Dist: protobuf==4.22.1; extra == "bigquery"
Requires-Dist: pyasn1==0.4.8; extra == "bigquery"
Requires-Dist: pyasn1-modules==0.2.8; extra == "bigquery"
Requires-Dist: python-dateutil==2.8.2; extra == "bigquery"
Requires-Dist: requests==2.28.2; extra == "bigquery"
Requires-Dist: rsa==4.9; extra == "bigquery"
Requires-Dist: six==1.16.0; extra == "bigquery"
Requires-Dist: urllib3==1.26.15; extra == "bigquery"
Provides-Extra: all
Requires-Dist: ingestra[postgres]; extra == "all"
Requires-Dist: ingestra[mysql]; extra == "all"
Requires-Dist: ingestra[bigquery]; extra == "all"
Dynamic: license-file

# 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

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

Run tests:

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

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

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

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

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

---

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