incorporator 1.1.3

Schema-free ETL mapper with in-process graph orchestration: turn any REST API, CSV, Parquet, or XML into typed Python objects, then coordinate multi-source pipelines in a bounded time window with Tideweaver.

---

🚀 Incorporator

A schema-free data mapper that turns JSON, XML, or CSV into a unified Python object graph with dot-notation and access-at-runtime — plus an in-process orchestrator (Tideweaver) for multi-source pipelines.

✨ Highlights

• Works with unpredictable JSON APIs — and digests XML, CSV, NDJSON, SQLite, Parquet without a line of schema. Native Python objects instantly, no manual model definitions.
• Handles changing structures at runtime — missing keys and mutating types absorbed without validation errors.
• Pydantic + HTTPX under the hood — no data classes, connection poolers, or pagination loops to write.

---

🛠️ How it Works: Zero-Schema Ingestion

Imagine this telemetry JSON. The nested "st" dictionary changes structure for every subsystem (pos vs sig vs bat). Standard parsers would crash.

[
  {"id":"NAV", "st":{"pos":[12,44], "ok":1}},
  {"id":"COM", "st":{"sig":78, "ok":1}},
  {"id":"PWR", "st":{"bat":92, "ok":1}},
  {"id":"THR", "st":{"lvl":63, "ok":0}}
]

Feed it the unpredictable JSON. Incorporator unifies the changing structures into a single object graph:

import asyncio
from incorporator import Incorporator

class System(Incorporator): pass     # Subclass; everything else hangs off it.

async def main():
    systems = await System.incorp(inc_file="telemetry.json", inc_code="id")
    print(systems.inc_dict["NAV"].st.pos)   # [12, 44]
    print(systems.inc_dict["PWR"].st.bat)   # 92

    thr = systems.inc_dict["THR"]
    if not thr.st.ok:
        print(f"⚠️ THRUST FAILURE! Efficiency dropped to {thr.st.lvl}")

asyncio.run(main())

The format is inferred from the URL or file extension. The syntax never changes for XML, CSV, NDJSON, SQLite, Parquet, Feather, ORC, Avro, or XLSX — same incorp() / export() surface.

---

📦 Installation

Built on Pydantic V2 metaprogramming, HTTPX, and Tenacity. No system dependencies.

pip install incorporator                  # core
pip install incorporator[speedups]        # orjson + lxml + cramjam
pip install incorporator[parquet]         # pyarrow — Parquet, Feather, ORC
pip install incorporator[avro]            # fastavro
pip install incorporator[orchestrate]     # typer + prefect — CLI + Prefect wrappers
pip install incorporator[all]             # everything except [parquet]

---

🧰 The Verbs + One Orchestrator

Every method you'll call on an Incorporator subclass, plus the windowed orchestrator.

incorp() — fetch, parse, build the object graph

class Launch(Incorporator): pass

launches = await Launch.incorp(inc_url="https://ll.thespacedevs.com/2.2.0/launch/upcoming/")
print(launches[0].name)

→ Tutorial 1 — First Steps

test() — let the framework write your incorp() kwargs

await Launch.test(inc_url="https://api.unknown.com/v1/users")
# Prints payload tree + suggested inc_code, rec_path, conv_dict.

refresh() — re-fetch live data into existing instances

await Launch.refresh(instance=launches)

The seed call's network context (params, headers, rec_path, conv_dict, ...) auto-replays.

export() — serialise to any format

await Launch.export(instance=launches, file_path="launches.parquet") — JSON, NDJSON, CSV, XML, SQLite, Parquet, Feather, ORC, Avro, XLSX. → Formats & compression

stream() — paginated bulk export, O(1) memory

For paginated APIs or local files too big for RAM, stream() fetches one page at a time, exports it, releases the page, and moves on — peak memory stays at roughly one chunk regardless of dataset size. A Wave per chunk is the built-in observability stream.

from incorporator.io.pagination import PageNumberPaginator

async for wave in Launch.stream(
    incorp_params={
        "inc_url": "https://api.example.com/launches",
        "inc_page": PageNumberPaginator(page_param="page"),
    },
    refresh_params=None,                                 # chunking: opt out of per-chunk refresh
    export_params={"file_path": "launches.ndjson", "if_exists": "append"},
):
    if wave.failed_sources: print(wave)

Use an append-friendly format (.ndjson / .csv / .sqlite / .avro) — Parquet, Feather, ORC, and Excel rebuild the whole file every wave and aren't safe for per-tick streams. For live dashboards keeping a registry hot across many sources, reach for fjord() instead.

→ Streaming & pagination · Tutorial 8

fjord() — a multi-source data pipeline

Fans out across N concurrent sources, fuses them through a user-defined outflow(state), exports the combined output.

async for wave in Incorporator.fjord(
    stream_params=[
        {"cls": Coin,  "incorp_params": {"inc_url": "..."}, "refresh_interval": 30},
        {"cls": Order, "incorp_params": {"inc_url": "..."}, "refresh_interval": 5},
    ],
    outflow="outflow.py",
    export_params={"file_path": "fusion.parquet"},
):
    if wave.failed_sources: print(wave)

→ Tutorial 10 — Multi-Source Fjord

Tideweaver — orchestrate multiple feeds on independent intervals

When you need several sources at different cadences inside a single time window, with dependency edges gating downstream work until upstreams produce fresh data, build a Watershed and hand it to Tideweaver:

from incorporator import Tideweaver, Watershed, Stream, Fjord

watershed = Watershed.diamond(
    window=(start, end),
    head=Stream(name="binance", cls=BinanceBook, interval=15, incorp_params={...}),
    middle=[Stream(name="coinbase", cls=CoinbaseTicker, interval=30, incorp_params={...}),
            Stream(name="kraken",   cls=KrakenTicker,   interval=30, incorp_params={...})],
    tail=Fjord(name="best_market", cls=BestMarket, interval=30,
               export_params={"file_path": "arb_signals.ndjson"}),
    outflow="arb_outflow.py",
)
async for tide in Tideweaver(watershed).run():
    print(tide.tide_number, tide.fired, tide.skipped)

Four shape helpers (parallel, chain, fanout, diamond) plus custom with explicit edges. Declarative watershed.json config + incorporator tideweaver run / validate CLI mirror the stream / fjord workflow.

→ Tutorial 11 — Tideweaver

When to reach for which long-running verb

Verb	Sources	Shape	Reach for it when…
stream()	one	paginated chunks, O(1) memory	bulk drain a paginated API or massive local file into a warehouse / archive
fjord()	many	stateful in-memory registry, live refresh	keep a hot multi-source object graph synchronised and snapshot it on a cadence
Tideweaver	many	windowed graph of streams + fjords with dependency edges	run several feeds at independent intervals inside a single time window, with downstream work gated on fresh upstream data

display() — REPL debug print: launches[0].display()

---

🚀 From Code to Production — CLI & Docker

The CLI runs the same engines from declarative config. No Python required.

Command	What it does
incorporator init --type {stream,fjord,tideweaver}	Scaffold a starter pipeline.json or watershed.json (+ outflow.py for fjord).
incorporator validate <config>.json	Structural check before you ship — no network calls. Auto-detects type.
incorporator stream pipeline.json	Run a single-source stream pipeline.
incorporator fjord pipeline.json	Run a multi-source fjord pipeline.
incorporator tideweaver run watershed.json	Run a windowed orchestration graph.

incorporator init --type stream --output-dir .
incorporator validate pipeline.json && incorporator stream pipeline.json   # ...or: docker compose up -d

Secrets stay out of config — ${API_KEY} for env vars, ${file:/run/secrets/api_key} for Docker / Kubernetes Secrets mounts. Set INCORPORATOR_SECRETS_ROOT=/run/secrets to sandbox ${file:...} against directory-traversal.

→ CLI reference · Deployment & secrets

---

🛠 Resilience & Batteries Included

• GIL-free hyperthreading via the [speedups] extra. → Installation
• Invisible decompression for .gz, .bz2, .lzma, .zip, .tar — ZIP/TAR paths validated against directory-traversal and a 1 GB bomb cap. → Formats
• Connection pooling + retries + DLQ — HTTP/2-multiplexed httpx.AsyncClient, Tenacity backoff, failed URLs on wave.failed_sources. Opt-in block_internal_redirects=True rejects 3xx Locations to RFC1918 / loopback / cloud-metadata IPs.
• Atomic writes + spreadsheet-injection guard — Parquet / Feather / ORC / JSON / XML / XLSX build via tempfile + os.replace() (no half-written files); CSV / XLSX cells starting with = / @ / + / - are quoted on export (OWASP).
• Non-blocking observability — subclass LoggedIncorporator; logs flow through a QueueHandler so disk I/O never blocks the event loop.
• Cross-format round-tripping — JSON ↔ Parquet ↔ SQLite ↔ Avro ↔ CSV ↔ XML. → Tutorial 3

---

📚 Tutorials (in order)

The eleven-tutorial curriculum. Each slot introduces one new verb or technique, alternating CoinGecko-heavy steps with non-CG domain examples so per-minute rate-limit windows refresh between CG calls and each Incorporator pattern lands across multiple real-world verticals. Runnable code under /examples.

1. 🌱 First Steps + DX Inspector — discovery-first flow: test() profiles a CoinGecko endpoint, then incorp() applies its recommendations.
2. 🔁 Data Lake Pivot — SaaS roster → BI-ready columnar; pivot a /users endpoint into Avro + SQLite.
3. 📦 Snapshot Warehouse — Universal Formats — fan CoinGecko top-100 snapshots into NDJSON / CSV / SQLite / Parquet, then round-trip every artifact.
4. 🛡️ XML Post Audit — federal-VIN fraud audit: XML invoice ledger enriched via one batched POST.
5. 🚀 Parent → Child Drilling — CoinGecko /coins/markets → /coins/{id} fan-out — the canonical backtest-data-prep pattern.
6. 🚀 SpaceX Launches — ops-dashboard feed: upcoming launches drilled for rocket + launchpad detail.
7. 🔄 Stateful Refresh — refresh() three ways against Binance's live ticker.
8. 🌊 Streaming Daemon — Paginated Bulk Export at O(1) Memory — stream()'s canonical job: chunking-mode drain of a paginated source. Plus a single-source stateful_polling=True compatibility shim with a pointer to T10 for the multi-source live-daemon path.
9. 🏁 NASCAR Fantasy Fjord — fantasy-sports scoring fjord across Cup, Xfinity, Truck series; previews T10's abstraction.
10. 🌊 Multi-Source Fjord — fjord() fusing CoinGecko + Binance into a live cross-venue spread metric.
11. 🧵 Tideweaver — Multi-Exchange Arb Scanner (capstone) — declarative windowed orchestration: three exchanges → one best-market record.

📑 Reference

• 📖 Library Reference — every public class, rendered from source docstrings.
• 📑 API Atlas — paste-ready map of every public callable: signature, pseudocode, "when to reach for it", common kwargs, tutorial cross-links.
• 🩺 Production Debugging with get_error() — LoggedIncorporator + structured error logs + DLQ retry.
• 📦 Formats & Compression + 🌊 Streaming & Pagination — every format kwarg, compression rules, and the paginator family for endpoints / files too big for RAM.
• 🐳 CLI & Configuration — running pipelines from pipeline.json / watershed.json.
• ⚡ Performance — measured throughput per format, memory profile, tuning knobs.

📎 Appendices — optional side-quests

• 🧬 Pokémon ETL — paginated HATEOAS drill + array reductions with calc / sum_attributes. Mirrors T5.
• 🕸️ Crypto Graph Mapping (static) — link_to-based in-memory join; T10's fjord pattern as a one-shot.
• 🏁 NASCAR Tideweaver — T11's diamond shape against race telemetry (laps + pits + flags → driver state).
• 🧵 Tideweaver Deep Dives — Parquet at window close and Tideweaver vs. Prefect — columnar artifacts plus the in-process-vs-cloud orchestration decision.

---

🤝 Philosophy & Contributing

See CONTRIBUTING.md for the dev install, quality bar, and architecture conventions. Security disclosures: SECURITY.md. Release notes: CHANGELOG.md.

---

Have a suggestion or hitting a snag? Edit this page on GitHub · Report an issue · Browse open issues