Metadata-Version: 2.4
Name: mini-atlas-graph-etl
Version: 0.2.0
Summary: GraphETL coordinator for GraphIngestor.
License-Expression: Apache-2.0
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Database
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: mini-atlas-graph-ingestor>=0.2.0
Requires-Dist: pydantic>=2
Requires-Dist: pyyaml
Requires-Dist: pyfiglet>=1.0
Provides-Extra: warehouse
Requires-Dist: mini-atlas-graph-warehouse>=0.1.1; extra == "warehouse"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Requires-Dist: setuptools>=77; extra == "dev"
Dynamic: license-file

# Mini Atlas Graph ETL

GraphETL is a thin **local/on-prem SQL-first orchestration layer**. Analysts
primarily provide SQL and declarative recipes. Local CSV files are a supported
**seed** source. 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 walkthroughs: [SQL expansion example](examples/sqlite_expansion/README.md),
[CSV seed example](examples/csv_seed/README.md).

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

## Installation

```bash
pip install mini-atlas-graph-etl
```

Optional GraphWarehouse publication:

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

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

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

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

```text
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, or a local CSV seed source
- 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

CSV-only recipes do not need connection factories or `--bindings`. Hybrid
recipes that seed from CSV and later expand in SQL still bind the SQL alias
only; a CSV step never consumes a connection.

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

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

Logging is **on by default** for the CLI (`--logging`). Progress and lifecycle
lines go to **stderr**; the rotating file log defaults to `logs/graph_etl/`
(`graph_etl_<UTC timestamp>.log`). Final JSON stays on **stdout**. `--verbose`
still only controls GraphIngestor bulk intervals. `--no-logging` skips the
banner, progress lines, and log file; concise fatal errors still print on
stderr and the exit code is unchanged.

Python callers stay silent until they opt in:

```python
from graph_etl import GraphETL, configure_logging, load_recipe

configure_logging(enabled=True, log_dir="logs/graph_etl")
recipe = load_recipe("recipe.yaml")
# Import runtime_bindings from your bindings module.
report = GraphETL(recipe, runtime_bindings).run()
```

Progress `batch` and `query_heartbeat` lines share the existing
`progress.report_every_seconds` interval (default 15). GraphETL does not log
per-node, per-edge, per-record, per-batch, or per-upsert successes. A failed
`run` writes one sanitized GraphETL wrapper traceback to the **file** log; the
console keeps the `run_failed` line plus the existing concise stderr error.
Logging does not change ETL results, checkpoints, lineage, or exports.

## Python API

```python
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; a seed may be SQL
  or a local CSV (`source.kind: csv`)
- one CSV can create multiple node and edge types through `ingestor_jobs`
- CSV seed followed by SQL expansion is supported; CSV expansion/reverse is not
- CSV connector values remain strings until existing normalization, Avro, or
  custom-normalizer layers
- `expansion` and `reverse` — the same frontier primitive with different labels;
  those steps remain SQL
- 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`.

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

```python
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 or NoSQL abstraction
- CSV is seed-only (no CSV expansion, reverse, Excel, URL, or compressed CSV)
- no CSV type map; values stay strings until existing normalization/Avro layers
- 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](examples/sqlite_expansion/README.md).

[CSV seed demo (no database, no `--bindings`)](examples/csv_seed/README.md).

GraphETL CSV **source ingestion** reads local CSV files as seed rows.
GraphWarehouse / GraphSlice CSV **export** writes a stored graph to CSV files.
They are different features.

## Development and release

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