Mini Atlas Graph ETL

GraphETL is a thin local/on-prem SQL-first orchestration layer. Analysts primarily provide SQL and declarative recipes. GraphETL runs seed, expansion, and reverse queries, feeds one in-memory GraphIngestor, and may publish a completed Parquet v1 export through GraphWarehouse.

GraphIngestor owns canonical identity, graph-schema ingestion, deduplication, integrity validation, and Parquet formatting. GraphETL does not reproduce that work. GraphWarehouse publication is optional. GraphQuery and GraphSlice remain outside GraphETL.

This repository does not include a UI, scheduler, workflow canvas, distributed execution, automatic connectors, credentials manager, or automatic supernode detection.

Runnable walkthrough: examples/sqlite_expansion/README.md.

Analyst how-to guides (short, task-oriented): docs/HOW_TO_INDEX.md.

Installation

pip install mini-atlas-graph-etl

Optional GraphWarehouse publication:

pip install 'mini-atlas-graph-etl[warehouse]'

The Mini Atlas Graph meta-package is the future one-command suite install:

pip install mini-atlas-graph

GraphETL can also be installed directly, as above. The distribution name is mini-atlas-graph-etl. The import package is graph_etl.

from graph_etl import GraphETL

Contributor editable installation from a clone (use python instead of python3 when that is what your system provides), after GraphIngestor 0.2.0 is already installable in the same environment:

pip install -e .
pip install -e ".[dev]"
pip install -e ".[warehouse]"

pip install -e ".[dev]" adds pytest for the tests in this repository. The optional extra name is mini-atlas-graph-etl[warehouse]. GraphWarehouse is not required to import GraphETL, validate recipes, preview rows, run Parquet-only jobs, or show CLI help.

Recipe versus runtime bindings

A recipe (YAML or JSON) is portable workflow configuration:

• SQL text or SQL-file paths
• seed / expansion / reverse order
• named frontiers
• normalization profile aliases
• GraphIngestor job mappings
• expansion bounds and hard caps
• output, checkpoint, lineage, and warehouse settings

Runtime bindings are environment-specific Python (RuntimeBindings):

• connection factories
• credentials loaded by user code, commonly from environment variables
• custom connector adapters
• custom batch normalizers
• runtime path overrides (output_directory, warehouse_db_path)
• live mapping or Avro dictionaries when you do not use recipe directories

Never put credentials in the recipe. GraphETL does not manage secrets.

The built-in connection-factory path is fixed:

• bind_style="named"
• max_bound_params=999

GraphETL does not infer DB-API paramstyle from the driver. Recipes have no binding-style setting. Drivers that need qmark, pyformat, or a different parameter cap use a custom adapter on RuntimeBindings.adapters.

CLI

graph-etl validate RECIPE [--bindings BINDINGS.py]
graph-etl preview RECIPE [--bindings BINDINGS.py] [--limit N]
graph-etl run RECIPE [--bindings BINDINGS.py] [--verbose] [--resume-from PATH]
graph-etl publish-warehouse EXPORT_DIR --db PATH [--run-id ID] [--force]

The same commands work as python3 -m graph_etl.cli ....

Python API

from graph_etl import GraphETL, load_recipe

recipe = load_recipe("examples/sqlite_expansion/recipe.yaml")
# Import runtime_bindings from your bindings module.
etl = GraphETL(recipe, runtime_bindings)
etl.validate()
rows = etl.preview(limit=10)
report = etl.run()

YAML and JSON

YAML is the documented primary format and is loaded with PyYAML. JSON recipes are also supported (.json). Both validate into the same Pydantic v2 models via load_recipe.

Query types

• seed — ordered as listed; multiple seeds are supported
• expansion and reverse — the same frontier primitive with different labels
• expansion/reverse consume a named single-column string frontier
• each consumer step keeps its own offset, so two steps can share a frontier without one consuming keys for the other

Frontier binding

SQL uses a literal {{frontier}} token. GraphETL replaces it with a parameterized IN (...) list. Values are chunked up to the connector’s max_bound_params. They are never interpolated as SQL literals.

The generic DB-API connector uses bound parameters and a configurable max_bound_params. Vendor-specific large-frontier strategies (for example temp tables) belong in a custom ConnectorProtocol adapter.

Custom adapters (qmark / pyformat)

RuntimeBindings.adapters maps a connection alias to an in-process ConnectorProtocol object stored by reference. A custom adapter wins over a factory for the same alias. This example uses SQLite’s qmark style and a smaller frontier chunk size. Copy it next to a recipe that already declares connection: source_a.

import sqlite3
from graph_etl import RuntimeBindings
from graph_etl.connectors import DbapiConnector

def connect_source() -> sqlite3.Connection:
    return sqlite3.connect("demo.sqlite")

runtime_bindings = RuntimeBindings(
    adapters={
        "source_a": DbapiConnector(
            connect_source,
            bind_style="qmark",
            max_bound_params=200,
        )
    }
)

A PostgreSQL-style driver that expects %(name)s placeholders declares bind_style="pyformat" the same way:

runtime_bindings = RuntimeBindings(
    adapters={
        "source_a": DbapiConnector(
            connect_source,
            bind_style="pyformat",
            max_bound_params=500,
        )
    }
)

connect_source is still user code: load DSN and credentials from the environment there. Do not put them in the recipe.

Normalization

The built-in conservative profile strips strings. Whitespace-only strings become None and increment values_set_null. Non-finite numbers and similar issues follow on_invalid. Custom batch normalizers are registered on RuntimeBindings.normalizers by profile name and may define different behavior.

normalization.max_reject_ratio is optional and normalization-only. It is a cumulative logical-run gate: the ratio is rejected_records / input_records across the whole run, including historical counts restored on resume. GraphETL evaluates it after each complete query-step invocation and again before integrity validation and Parquet export. Fetch batch size does not change the outcome. The comparison is strict > rather than >=, so a ratio equal to the threshold succeeds. drop_record and quarantine count toward rejected_records; set_null does not. GraphIngestor per-job failed counts still appear in RunReport.counts and lineage. They are not added to this ratio: one source row can feed several ingestion jobs, so summing job-level failures could double-count a row and produce a ratio greater than one. v0.1 does not add a separate ingestion-failure threshold.

GraphIngestor still owns final property mapping, Avro coercion/validation, canonical identity, and graph deduplication.

GraphIngestor orchestration

• one GraphIngestor per run
• node jobs run before dependent edge jobs
• edge jobs declare raw from_column / to_column plus required from_label / to_label
• GraphETL calls public GraphIngestor.canonical_node_id for those endpoints
• validate_integrity is mandatory before the final Parquet export

A GraphETL run may still complete successfully when GraphIngestor rejects individual records. Those rejections are not part of max_reject_ratio. When any GraphIngestor failed count is nonzero, RunReport, lineage, and CLI JSON include a warning such as GraphIngestor rejected N records across M jobs plus top-level ingestor_rejected_records. The warning does not include rejected rows, identifiers, property values, SQL, or callable representations.

A successful run status does not mean every source row was stored. Use ingestor_rejected_records, per-job *_failed counts, and the warning above. v0.1 has no GraphIngestor-failure threshold that flips status.

Raw source IDs are opaque. A colon in a value such as a URL, URN, timestamp, or namespaced ID does not mean the value is already canonical. GraphETL passes raw endpoint columns to GraphIngestor; GraphIngestor composes canonical IDs.

Do not reproduce GraphIngestor identity logic or call _create_node_id.

Output and optional warehouse publication

Each run writes a unique Parquet v1 directory:

• metadata.json, nodes/, and edges/ come from GraphIngestor
• etl_lineage.json is a GraphETL sidecar written after the warehouse attempt (including skip or failure)

When warehouse.enabled is true and the [warehouse] extra is installed, GraphETL calls GraphWarehouse.ingest_run. A warehouse failure sets status warehouse_publication_failed and preserves the Parquet export. That failure wins over a late cancellation. graph-etl publish-warehouse retries publication only; it does not rerun SQL or GraphIngestor.

Republishing the same resolved export path with the same run_id and a compatible graph_id is a successful skip (skipped: true). Compatible means equal graph IDs when both are present, or both missing. Present versus absent, or two different graph IDs, is a conflict. That skip is an identity match only; it is not proof that export bytes are unchanged. Reusing a run_id for a different resolved path or an incompatible graph_id is an identity conflict: publication fails and the CLI returns 1. --force (or skip_if_ingested=False) reingests by upsert; it does not replace every row previously contributed by that run.

Cancellation before publication prevents it. A completed successful or skipped Warehouse result is kept even if cancellation arrives afterward.

Progress, cancellation, and limits

• periodic elapsed-only heartbeats while a source execute() is blocked
• cooperative cancellation between batches
• GraphETL does not promise to kill an arbitrary blocked database call
• max_rounds and stop-when-no-new-keys are normal completion
• hard record/frontier caps return status="limit_reached" with a valid partial Parquet export; there is no public FrontierLimitError
• warehouse publication after a hard cap follows warehouse.publish_on_limit (default false)
• expansion/reverse consumers keep independent offsets, so two steps can share a frontier without one consuming keys for the other
• expansion.stoplist_frontiers names frontiers whose keys are omitted from later binds; that is a configured-key guardrail, not automatic degree/supernode detection
• frontier IN (...) lists are chunked up to the connector’s max_bound_params (factory path: 999; adapters may set another cap)

Checkpoint and resume

checkpoint.policy:

• none (default) — no frontier checkpoint
• until_success — checkpoint until a successful export, then remove owned staging
• keep — leave the checkpoint for later resume

Safe checkpoints are written after seeds or complete expansion rounds. Checkpoint format v2 uses 11 internal SQLite tables. Resume restores cumulative processed/normalization/quarantine counts, frontier logs and per-consumer offsets, GraphIngestor job aggregates, and the logical start time. Ordered seeds and frontier consumers are validated before source access.

A later run that finds owned checkpoint staging and has no explicit resume fails before source access. Continue with graph-etl run --resume-from PATH or resume_from=. That path can load a permission-restricted snapshot.

Resume checks a semantic fingerprint of the recipe plus resolved execution inputs: SQL text (inline or sql_file contents), resolved property mappings and Avro configuration, and deterministic runtime binding identity (module and qualified name, or CLI bindings path digest). Fingerprinting excludes progress.verbose, output.directory, and warehouse.db_path. Callable bodies are not fingerprinted. Changing live source data or the implementation of a same-identity Python callable is not detected.

GraphETL does not pickle the graph. Owned checkpoint staging directories use mode 0o700 and owned snapshot/SQLite files use 0o600 where the OS supports it. Unsupported chmod, including Windows NotImplementedError, does not fail the run. Final user export directories, source SQL, runtime bindings files, Warehouse databases, and other user inputs are not chmod’d.

Security and governance

• no credentials in recipes
• bound parameters instead of SQL interpolation
• redacted errors, logs, and lineage ([REDACTED]); public exception chains keep safe type and category information and do not retain original secret-bearing exception objects
• lineage stores counts, aliases, and SQL hashes or templates — not passwords, full DSNs, bound values, frontier keys, or raw rows

Current intentional limitations

• the GraphIngestor graph stays in memory until export
• v1 frontiers are single-column strings
• no universal vendor connector, CSV connector, or NoSQL abstraction
• no credentials manager and no recipe-level bind_style
• no GraphIngestor-failure threshold
• no automatic supernode / high-degree detection (stoplists are configured keys)
• no mandatory source-query timeout
• no workflow scheduler, workflow canvas, or distributed workers
• no UI in this repository
• GraphQuery and GraphSlice are downstream of a populated warehouse and are outside GraphETL’s runtime scope

Example

Domain-neutral SQLite seed / expansion / reverse demo.

Development and release

Local test commands and private-repository release gates are in RELEASE_CHECKLIST.md. Public package-index publication is not required.