Metadata-Version: 2.5
Name: datadongle
Version: 0.1.1
Summary: Installable data-collector tooling: source collectors, pluggable storage engines, and staged-ingest/SCD2 load strategies.
Project-URL: Homepage, https://github.com/matttriano/datadongle
Project-URL: Repository, https://github.com/matttriano/datadongle
Project-URL: Documentation, https://github.com/matttriano/datadongle#readme
Project-URL: Issues, https://github.com/matttriano/datadongle/issues
Author: Matt Triano
License-Expression: Apache-2.0
License-File: LICENSE
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Database
Classifier: Topic :: Scientific/Engineering :: GIS
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.13
Requires-Dist: ijson>=3.5.0
Requires-Dist: numpy
Requires-Dist: openpyxl>=3.1.5
Requires-Dist: pandas
Requires-Dist: pyyaml
Requires-Dist: requests
Requires-Dist: tenacity
Provides-Extra: all
Requires-Dist: duckdb; extra == 'all'
Requires-Dist: fiona; extra == 'all'
Requires-Dist: geopandas; extra == 'all'
Requires-Dist: networkx; extra == 'all'
Requires-Dist: psycopg2-binary; extra == 'all'
Requires-Dist: pyarrow; extra == 'all'
Requires-Dist: pyiceberg[sql-sqlite]; extra == 'all'
Requires-Dist: pymysql; extra == 'all'
Requires-Dist: rasterio; extra == 'all'
Requires-Dist: shapely; extra == 'all'
Provides-Extra: geo
Requires-Dist: fiona; extra == 'geo'
Requires-Dist: geopandas; extra == 'geo'
Requires-Dist: networkx; extra == 'geo'
Requires-Dist: rasterio; extra == 'geo'
Requires-Dist: shapely; extra == 'geo'
Provides-Extra: iceberg
Requires-Dist: duckdb; extra == 'iceberg'
Requires-Dist: pyarrow; extra == 'iceberg'
Requires-Dist: pyiceberg[sql-sqlite]; extra == 'iceberg'
Provides-Extra: mysql
Requires-Dist: pymysql; extra == 'mysql'
Provides-Extra: postgres
Requires-Dist: psycopg2-binary; extra == 'postgres'
Description-Content-Type: text/markdown

# datadongle

[![CI](https://github.com/matttriano/datadongle/actions/workflows/ci.yml/badge.svg)](https://github.com/matttriano/datadongle/actions/workflows/ci.yml)
[![PyPI version](https://img.shields.io/pypi/v/datadongle.svg)](https://pypi.org/project/datadongle/)
[![Python versions](https://img.shields.io/pypi/pyversions/datadongle.svg)](https://pypi.org/project/datadongle/)
[![License: Apache-2.0](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://github.com/matttriano/datadongle/blob/main/LICENSE)

Installable data-collector tooling: **source collectors**, **pluggable storage engines**, and **staged-ingest / SCD2 load strategies** — decoupled so you can mix and match.

A collector says *what* to pull (a Socrata dataset, say). A **write mode** says *how* new rows should reconcile with what's already stored (append, upsert, or keep versioned history). A **storage engine** decides *where* and *physically how* that happens (Postgres/PostGIS or a local Iceberg warehouse). These three axes are independent: the same collector runs unchanged onto either engine, under any compatible write mode.

```
   SourceReader           WriteMode            Engine
  (what to collect)   (how to integrate)   (where it lands)
        │                    │                   │
 SocrataReader ──▶  Append / Upsert / SCD2  ──▶  PostgresEngine
                                                 IcebergEngine
```

---

## Installation

datadongle targets **Python ≥ 3.13**. The base install is deliberately lean; storage backends and geo support are optional extras — install only what you need.

| Extra | Pulls in | Needed for |
|-----------|-------------------------------------------------|-----------------------------------------------|
| `postgres` | `psycopg2-binary` | `PostgresEngine` |
| `mysql` | `pymysql` | `MySQLEngine` |
| `iceberg` | `pyiceberg[sql-sqlite]`, `duckdb`, `pyarrow` | `IcebergEngine` |
| `geo` | `shapely`, `geopandas`, `fiona`, `rasterio`, … | geometry columns on **either** engine |
| `all` | all of the above | everything |

For end users installing from PyPI:

```bash
pip install datadongle                    # base
pip install "datadongle[postgres]"        # Postgres storage engine
pip install "datadongle[iceberg,geo]"     # Iceberg + geometry support
pip install "datadongle[all]"             # everything
```

For local development with [`uv`](https://docs.astral.sh/uv/) (see [Contributing](#contributing)):

```bash
# Postgres target with geometry support
uv sync --extra postgres --extra geo

# Local Iceberg target with geometry support
uv sync --extra iceberg --extra geo

# everything (all extras + dev tools)
uv sync --all-extras
```

`IcebergEngine` uses **no DuckDB native extensions** — PyIceberg does all Iceberg I/O, core DuckDB does the change-detection join, and shapely handles WKB geometry. It runs fully offline against a local-filesystem warehouse.

---

## Quickstart

Collect a Socrata dataset into a local Iceberg warehouse. This example is self-contained (no database to stand up):

```python
from datadongle.collectors.socrata.reader import SocrataReader
from datadongle.collectors.socrata.spec import SocrataDatasetSpec
from datadongle.engines.iceberg import IcebergEngine
from datadongle.load.driver import run_collection

spec = SocrataDatasetSpec(
    name="chicago_building_permits",
    dataset_id="ydr8-5enu",       # Socrata 4x4 id
    target_table="building_permits",
    target_schema="raw_data",
    entity_key=["permit_"],       # non-empty entity_key ⇒ SCD2 history
)

reader = SocrataReader()                       # optional: app_token=..., page_size=...
engine = IcebergEngine("/data/warehouse")      # local warehouse directory

# First load: read everything.
run_collection(reader, spec, engine, mode="full")

# Later runs: read only what changed since the last load.
summary = run_collection(reader, spec, engine, mode="incremental")
print(summary)
# {'source': 'socrata', 'dataset_id': 'ydr8-5enu', 'mode': 'incremental',
#  'rows_staged': 42, 'rows_merged': 3, 'rows_invalidated': 0, 'high_water_mark': ...}

# Query the result (current version of each permit, geometry parsed to shapely):
target = reader.target(spec)
current = engine.read_current(target)
```

Point the same collection at Postgres instead — nothing else changes:

```python
from datadongle.db.core import DatabaseCredentials
from datadongle.engines.postgres import PostgresEngine

creds = DatabaseCredentials(
    host="localhost", port=5432, database="dwh",
    username="etl", password="…",
)
engine = PostgresEngine(creds)

run_collection(reader, spec, engine, mode="full")
```

`run_collection` returns a summary `dict` with `rows_staged`, `rows_merged`, `rows_invalidated`, and the resulting `high_water_mark`. Pass an optional `tracker` to record each run for observability — it is **not** the source of truth for incremental resumption (see below).

---

## Collection modes: `full` vs `incremental`

The **collection mode** controls *how much* of the source to read on a given run. It is chosen per-call via `run_collection(..., mode=...)` (default `"incremental"`).

- **`full`** — read the entire source (`since=None`). Use for the first load, for full refreshes, and whenever the source isn't incrementally queryable.

- **`incremental`** — read only rows newer than what's already stored. The driver asks the engine for the target table's **high-water mark** (the max cursor value, e.g. `max(socrata_updated_at)`), and the reader turns that into a source-side filter.

The high-water mark is read **from the target table itself**, never from a run log:

```python
engine.read_high_water_mark(target, cursor_spec)   # max(cursor) + tiebreak, in the engine's dialect
```

This is deliberately **self-healing**: drop and rebuild the table and the next incremental run automatically restarts from the correct point, because the mark lives with the data. A tracker, if supplied, records the mark only for observability.

If the source has no cursor (`reader.cursor_spec(spec)` returns `None` — e.g. a Socrata `file_download` export, which carries no system fields), an `incremental` request transparently falls back to a full read.

> **Timestamps are UTC.** Both engines store `TIMESTAMPTZ` columns as UTC instants and pin their session/connection to UTC, so high-water marks round-trip identically regardless of the host or server timezone.

---

## Write modes: `Append`, `Upsert`, `SCD2`

The **write mode** is the *policy* for reconciling incoming rows with the target — independent of the engine, which supplies the *mechanism*. A collector selects a policy without knowing the storage. `SocrataReader`, for instance, returns `SCD2(entity_key=...)` when the spec has an `entity_key`, otherwise `Append`.

```python
from datadongle.core.write_mode import Append, Upsert, SCD2
```

### `Append()`
Insert every incoming row. No key, no deduplication, no versioning — the target accumulates everything it's given, duplicates included. Good for immutable event/log data.

### `Upsert(keys, on_conflict="update")`
Insert-or-update keyed by `keys`.
- `on_conflict="update"` — overwrite the conflicting row's non-key columns from the incoming row (last write wins).
- `on_conflict="nothing"` — keep the existing row, ignore the incoming duplicate.

Keeps exactly one row per key; **no history**. Supported by both engines (`IcebergEngine` uses PyIceberg's native `upsert`).

### `SCD2(entity_key, invalidate_missing=False)`
Keep **versioned history** keyed by `entity_key` plus a content hash. A new version is written **only when an entity's content actually changes**:

- The engine computes a `record_hash` over the entity's *data* columns, **excluding** the `entity_key` (identity, not content) and any **metadata columns** (e.g. Socrata's `socrata_id` / `socrata_updated_at`, which change every run regardless of content).
- An unchanged re-pull is a **no-op** — same hash ⇒ no new version.
- A metadata-only change (e.g. a bumped `updated_at` with identical data) does **not** create a version.
- A genuine data change appends a new version and the entity's "current" pointer moves to it.

`invalidate_missing=True` additionally closes out / tombstones entities that are **absent** from the pull. Because "absent" can only be judged against a complete snapshot, this requires `mode="full"` — the driver raises if you request it incrementally.

**How each engine realizes SCD2:**

| | `PostgresEngine` | `IcebergEngine` (Shape B) |
|--------------------|------------------------------------------------|-----------------------------------------------|
| Physical shape | `valid_from` / `valid_to` columns updated in place | Append-only satellite; **no** `valid_to` |
| "Current" version | `WHERE valid_to IS NULL` | Derived at read time: latest `effective_from` per `entity_key` (window function) |
| Version columns | `record_hash`, `valid_from`, `valid_to` | `record_hash`, `effective_from`, `ingested_at`, `load_id` |
| Integrity | unique index on `(entity_key, record_hash)` + partial index for current | dedupe via DuckDB anti-join against history |
| `invalidate_missing` | sets `valid_to` on vanished entities | appends a tombstone version (sentinel hash), hidden from current |

Both engines yield the **same logical outcome** — identical row counts, the same no-op/version decisions, the same current-state — verified by the two-engine conformance suite (`tests/engines/test_conformance.py`).

---

## Storage engines

Both engines implement the same `Engine` protocol (`ensure_table`, `open_write`, `query`, `read_high_water_mark`, `table_columns`, `geometry_columns`, …), so they are interchangeable under `run_collection`.

### `PostgresEngine(creds, db_name=None, *, manage_ddl=True)`
Postgres + PostGIS. `ensure_table` renders `CREATE TABLE IF NOT EXISTS` DDL (geometry columns become `geometry(<kind>,<srid>)`); writes go through a `COPY`-into-staging then per-mode merge (`append_merge` / `upsert_merge` / `scd2_merge` in `engines/postgres_load.py`). `query(...)` returns a `DataFrame`, or a `GeoDataFrame` when a PostGIS geometry column is present. Needs the `postgres` extra (and `geo` for geometry). See [Version-controlled DDL](#version-controlled-ddl) for `manage_ddl`.

### `IcebergEngine(warehouse, catalog_name="datadongle")`
A local-filesystem Iceberg warehouse (PyIceberg + a SQLite catalog) queried through DuckDB. Geometry is stored as WKB `binary` with the SRID retained in table properties. Shape-B SCD2 keeps writes cheap (pure appends). Reads:

```python
engine.read_current(target)     # latest version per entity  → (Geo)DataFrame
engine.read_history(target)     # every stored version       → (Geo)DataFrame
engine.query("select … from <table>_current where …")   # DuckDB SQL; <table> and <table>_current views registered
```

Needs the `iceberg` extra (and `geo` for geometry). No native DuckDB extensions required.

---

## Version-controlled DDL

By default `PostgresEngine` creates its own tables. If your schema is owned by a migration tool (Flyway, sqitch, a checked-in SQL script), you want the opposite: datadongle should *describe* the table it needs and let the migration tool apply it, so an ingestion run can never create a table your migration history has no record of.

### Get the DDL

`render_create_table` is a pure function of `(TableRef, TableSchema, WriteMode)` — no connection, no credentials:

```python
from datadongle.engines.postgres_ddl import render_create_table

print(render_create_table(reader.target(spec), reader.schema(spec), reader.write_mode(spec)))
```

```sql
create table raw_data.chicago_building_permits (
  "permit_" text not null,
  "issue_date" timestamptz,
  "geom" geometry(Point,4326),
  "ingested_at" timestamptz not null default (now() at time zone 'UTC'),
  "record_hash" text not null,
  "valid_from" timestamptz not null default (now() at time zone 'utc'),
  "valid_to" timestamptz
);

create unique index uq_chicago_building_permits_entity_hash
    on raw_data.chicago_building_permits ("permit_", "record_hash");

create index ix_chicago_building_permits_current
    on raw_data.chicago_building_permits ("permit_") where "valid_to" is null;
```

Note what a hand-written migration would have missed: `ingested_at` on every table, the SCD2 versioning trio, and two indexes the merge SQL depends on. That is why this is generated rather than transcribed.

Output is bare DDL — a versioned migration runs exactly once, so an object that already exists should fail loudly. Pass `if_not_exists=True` for a Flyway repeatable (`R__`) migration, and `include_schema=True` to prepend `create schema if not exists <namespace>;`.

Paste it into `V1__create_chicago_building_permits.sql` and run `flyway migrate`.

### Hand over schema ownership

```python
engine = PostgresEngine(creds, manage_ddl=False)
run_collection(reader, spec, engine, mode="full")
```

`ensure_table` now executes no DDL. It asserts the table exists and matches the collector's schema, raising `TableNotFoundError` (with the `create table` to apply) or `SchemaDriftError` (with the `alter table` to apply) instead of quietly creating or ignoring.

### Handle drift

When an upstream source adds a field, the next run fails with the migration you need rather than silently dropping the column:

```python
engine.diff_table(target, schema, mode)      # SchemaDiff: missing / unexpected / retyped
engine.render_migration(target, schema, mode)
# alter table raw_data.chicago_building_permits add column "applicant_name" text;
```

`render_migration` only handles **additive** drift. A dropped or retyped column raises instead, because resolving it needs a decision about existing rows that datadongle can't make for you — at a raw ingestion layer, writing to a new table version is usually safer than an in-place change. A `not null` column is added nullable with the constraint emitted as a commented-out follow-up, since `ADD COLUMN … NOT NULL` fails on a populated table.

`IcebergEngine` is unaffected: it creates tables through the PyIceberg catalog API rather than SQL DDL, and has native schema evolution.

---

## Contributing

Contributions are welcome. This project uses [`uv`](https://docs.astral.sh/uv/) for dependency management; a dynamic version derived from git tags via [hatch-vcs](https://github.com/ofek/hatch-vcs) (there is no version string to edit).

```bash
git clone https://github.com/matttriano/datadongle
cd datadongle
uv sync --all-extras        # installs the project, all extras, and dev tools
```

Before opening a pull request:

```bash
uv run ruff format .        # format
uv run ruff check .         # lint
uv run ty check             # type-check
uv run pytest               # tests (network + live-DB tests deselected by default)
```

CI runs formatting, linting, type-checking, tests, a build check, and security scans (gitleaks, zizmor, pip-audit) on every pull request. Changes under `.github/` require review from a code owner.

---

## Testing

```bash
uv run pytest                      # hermetic tests (Iceberg + unit); network + DB tests skip
uv run pytest -m network           # opt in to the network-marked tests
uv run pytest -m postgres          # opt in to the live-Postgres tests (see below)
```

- **Iceberg tests are hermetic** — they build a warehouse under a `tmp_path`, so they run anywhere with no external service.
- **Postgres-backed tests skip** unless a database is configured. Set `DWH_TEST_PGHOST`, `DWH_TEST_PGPORT`, `DWH_TEST_PGDATABASE`, `DWH_TEST_PGUSER`, `DWH_TEST_PGPASSWORD` and they light up — including the Postgres arm of the two-engine conformance suite:

  ```bash
  DWH_TEST_PGHOST=localhost DWH_TEST_PGPORT=5432 \
  DWH_TEST_PGDATABASE=dwh_test DWH_TEST_PGUSER=postgres DWH_TEST_PGPASSWORD=… \
  uv run pytest -m postgres tests/engines/test_conformance.py
  ```

- **Network-marked tests are deselected by default** (they need egress); run them explicitly with `-m network`.

---

## Releasing

Versions are derived from git tags — there is no version string to edit. Every merge to `main` publishes an auto-versioned dev build (`X.Y.Z.devN`) to TestPyPI; a `v*` tag publishes a clean release to PyPI via Trusted Publishing (no stored tokens).

### Rehearse on TestPyPI

Merges to `main` publish to TestPyPI automatically. To verify an install from there (pulling real dependencies from PyPI, since TestPyPI doesn't host them):

```bash
uv run --no-project --with datadongle \
  --index https://test.pypi.org/simple/ \
  --extra-index-url https://pypi.org/simple/ \
  -- python -c "import datadongle; print(datadongle.__version__)"
```

### Publish a release to PyPI

Confirm `main` is green and the TestPyPI dev build looks right, then tag:

```bash
git checkout main
git pull origin main
git tag -a v0.1.0 -m "Release 0.1.0"
git push origin v0.1.0
```

The tag triggers the release workflow, which runs tests, then publishes to PyPI after a required-reviewer approval. PyPI versions are **write-once** — to fix a broken release, bump the version and tag again.
