datadongle

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.

For end users installing from PyPI:

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 (see Contributing):

# 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):

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:

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:

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.

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:

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

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:

from datadongle.engines.postgres_ddl import render_create_table

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

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

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:

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 for dependency management; a dynamic version derived from git tags via hatch-vcs (there is no version string to edit).

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:

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

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:

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

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:

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.