Metadata-Version: 2.4
Name: mini-atlas-graph-ingestor
Version: 0.2.0
Summary: Build validated in-memory property graphs for ETL persistence and analysis.
License-Expression: Apache-2.0
Keywords: etl,graph,parquet,cli
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: fastavro
Requires-Dist: pyarrow
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Dynamic: license-file

# Mini Atlas Graph Ingestor

In-memory graph construction for ETL jobs. Query, clean, and normalize rows
in the ETL, then pass already-clean records through `GraphIngestor` to build
nodes and edges and export Parquet that GraphWarehouse can load.

This README is the ETL-oriented usage guide. SQL-first orchestration lives
in GraphETL, a separate package.

Public import:

```python
from graph_ingestor import GraphIngestor
```

`LoggerHelper` and every method whose name starts with `_` are internal.
Do not call them from ETL code.

## 1. Purpose and scope

`GraphIngestor` validates already-clean records, applies deterministic upsert
rules, keeps an in-memory property graph, and writes a trusted pickle or
GraphWarehouse Parquet export.

**Belongs in the upstream ETL**

- SQL (or other) queries
- cleaning, trimming, case folding, and domain-specific type coercion
- assigning timezones (this library never implies one)
- producing scalar ids and property values

**Belongs in GraphWarehouse**

- durable warehouse storage and analysis after Parquet export
- this repository only defines the interchange directory layout and columns

**Non-goals**

- no database access
- no pandas
- no implied timezones or string cleaning
- `weighted = True` does not create a weight property
- no adjacency index
- no context-manager / `close` API
- no public logging API beyond constructor `enable_logging` / `log_level`
  and bulk `log_every`

## 2. Installation and supported Python versions

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

For local development:

```bash
pip install -e ".[dev]"
```

Requires Python 3.10+. Runtime dependencies are `fastavro` and `pyarrow`.
Classifiers currently list 3.10–3.13. Development extra is `pytest`.

`upsert_bulk_nodes` / `upsert_bulk_edges` accept `show_progress = True`. If
`tqdm` is installed, rows wrap in a progress bar that is closed after success
or an ordinary exception; if tqdm is not installed, progress is silently
skipped. `tqdm` is not a declared extra. A real `progress` extra is a possible
later enhancement.

Keyword-only `log_every` (default `100_000`) controls INFO progress during
bulk ingest. `None` disables intermediate progress; the start line and final
summary still log at INFO. `0`, negative values, booleans, and other non-
integers raise `ValueError` before any rows are processed.

Entry points after install:

```bash
graph-ingestor --help
python -m graph_ingestor --help
```

## 3. Graph data model

### Node records

A node is stored under its canonical id. Default root keys:

```python
{
    "uid": "person:123",
    "type": "person",
    "properties": {"name": "A. Person"},
    "event_ts": "2026-08-24T20:00:00+00:00",  # optional
}
```

Constructor options rename the in-memory root keys (`node_id_key_name`,
`label_key_name`, `property_key_name`). Parquet interchange columns stay
`uid`, `type`, `properties`, `event_ts`.

### Edge records

An edge id is `(from_id, label, to_id)`. Default stored record:

```python
{
    "from_id": "person:123",
    "to_id": "company:456",
    "type": "works_at",
    "properties": {"role": "engineer"},
    "event_ts": "2026-08-24T20:00:00+00:00",  # optional
}
```

### Canonical identities

- Node id: `<label>:<source-id>`, for example `email_address:john@gmail.com`.
  `canonical_node_id(label, source_id)` returns that string without inserting
  a node. It is the same helper used by `create_node_object`, `ingest_node`,
  and default `upsert_bulk_nodes`.
- `create_node_object` / `ingest_node` accept a source id and canonicalize it.
- Default `upsert_bulk_nodes` treats each row id as an **opaque source
  value**. A colon is not proof of canonical form. URLs, URNs, `12:30`,
  `ns:id`, and a raw value that already begins with the bulk label are all
  composed as `label:<normalized source>`, matching `canonical_node_id`.
- Pass `ids_are_canonical=True` only when every row id is already
  `<label>:<source-id>` for that bulk label. A mismatched prefix fails that
  row. This is a breaking change for callers that used prefixed uids as bulk
  source ids without the flag.
- `upsert_node_object`, `remove_node`, and `purge_nodes` require the canonical
  id.
- Labels must be safe Parquet partition filenames (see Label safety). `:` is
  reserved as the identity separator and is never allowed in a label.
- Source ids must be non-null scalars with a non-empty string form. Bytes and
  collections are rejected.
- Whitespace in the source-id portion is replaced with
  `id_whitespace_replacement` (default `_`), and non-string scalars are
  converted with `str()`. Nothing else is cleaned or case-folded. An empty
  replacement string becomes `_`. Inputs that normalize to the same canonical
  value are the **same source identity** by contract. Existing GraphWarehouse
  databases keep this format.

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(weighted = False, directional = True, validate_avro = False)
person = ingestor.create_node_object("person", "Ada Lovelace", {"name": "Ada"})
assert person["uid"] == "person:Ada_Lovelace"
email = ingestor.create_node_object("email_address", "john@gmail.com", {})
assert email["uid"] == "email_address:john@gmail.com"
assert ingestor.canonical_node_id("person", "http://example.com/a") == (
    "person:http://example.com/a"
)
assert ingestor.upsert_bulk_nodes(
    [{"uid": "urn:uuid:1234", "properties": {}}],
    "person",
)["created"] == 1
assert "person:urn:uuid:1234" in ingestor.nodes
assert ingestor.upsert_bulk_nodes(
    [{"uid": "person:1", "properties": {}}],
    "person",
    ids_are_canonical = True,
)["created"] == 1
```

### Label safety

Node and edge labels become Parquet filenames under `nodes/` and `edges/`.
They must remain a single path name on macOS, Linux, and Windows:

- rejected: empty or whitespace-only strings
- rejected: `:` (canonical identity separator)
- rejected: path separators `/` and `\`
- rejected: filename-unsafe characters `<` `>` `"` `|` `?` `*`
- rejected: `.` and `..`
- rejected: Unicode category `Cc` control characters
- rejected: labels that would not remain a single POSIX path name
- rejected: labels ending in `.` or whitespace
- rejected: Windows reserved device basenames, matched case-insensitively on
  the token before the first `.`: `CON`, `PRN`, `AUX`, `NUL`, `COM1`–`COM9`,
  `LPT1`–`LPT9`

Allowed: mixed case, hyphens, underscores, internal spaces, Unicode letters
or numbers, and names such as `COM10`, `CONSOLE`, and `NULLED`.

Labels stay case-sensitive in the graph. Parquet export detects filename
collisions before writing: `Person` versus `person`, and NFC versus NFD
equivalents of the same letters, fail with `ValueError` naming the
conflicting labels. Node and edge partition directories are checked
separately. Nothing is written, and an existing destination is left
unchanged.

```python
from graph_ingestor import GraphIngestor

g = GraphIngestor(weighted = False, directional = True, validate_avro = False)
g.create_node_object("email_address", "john@gmail.com", {})
try:
    g.create_node_object("CON", "x", {})
    raise AssertionError("Windows reserved label should fail")
except ValueError:
    pass
try:
    g.create_node_object("a:b", "x", {})
    raise AssertionError("colon in label should fail")
except ValueError:
    pass
```

### Root fields versus properties

Root identity, label, property-container, endpoint, and `event_ts` keys must
not also appear inside `properties`. That raises `ValueError`.

### JSON-safe properties

Property values must be representable as strict JSON before commit, on single
and bulk node and edge paths, with or without Avro. Accepted: `None`,
booleans, integers, finite floats, strings, lists, and dictionaries with
string keys, including nested structures. Shared non-circular lists or dicts
referenced from multiple parents are valid. Actual circular references,
`NaN`, `+Infinity`, `-Infinity`, `bytes`, `set`, `datetime`, non-string
dictionary keys, and other custom values are rejected. Ingest uses a cheap
recursive walk (objects on the current stack only; no serialized copies).
Parquet export uses `json.dumps(..., allow_nan=False)` as defense in depth.
Avro coercion does not make non-finite numbers acceptable. Validation and
copying happen in one pass: accepted nested structures are copied while they
are checked, without a serialized cache.

### `event_ts`

Optional on create/upsert/ingest. Accepted: timezone-aware `datetime`,
timezone-aware ISO-8601 string, or `None`. Stored as a UTC ISO-8601 string at
record root, never inside `properties`. Naive datetimes and malformed strings
raise `ValueError`. A non-null incoming `event_ts` on upsert replaces the
existing root value; `date_keys` do not apply to it. Compatible GraphWarehouse
versions store the exported offset-aware strings as naive UTC DuckDB
`TIMESTAMP`.

### Directed versus undirected

`directional = True`: endpoint order is significant.
`directional = False`: endpoints are ordered lexicographically, so
`(a, label, b)` and `(b, label, a)` are the same edge. Self-edges are allowed.

### Weighted versus unweighted

`weighted` is graph metadata only. A required weight belongs in the caller's
schema and properties.

`nodes`, `edges`, `node_types`, `edge_types`, and the count attributes remain
visible for compatibility. Treat them as read-only. Mutating them directly is
unsupported.

## 4. Constructor options

```python
GraphIngestor(
    weighted: bool,
    directional: bool,
    property_mappings = None,          # dict[label, {source_key: target_key}]
    node_id_key_name = "uid",
    edge_from_id_key_name = "from_id",
    edge_to_id_key_name = "to_id",
    label_key_name = "type",
    property_key_name = "properties",
    date_keys = None,                  # property -> min|before|max|after
    incremental_keys = None,           # property -> add|subtract
    validate_avro = True,
    id_whitespace_replacement = "_",
    enable_logging = False,
    log_dir = "logs",                  # str | PathLike[str]
    log_file = "graph_ingestor.log",
    log_level = logging.INFO,
    graph_name = "",
    graph_description = "",
)
```

| Parameter | Default | When to set it |
|---|---|---|
| `weighted` | required | Metadata flag only |
| `directional` | required | `False` only for undirected graphs |
| `property_mappings` | `{}` | Inline maps, or load files later |
| `node_id_key_name` | `"uid"` | Match bulk node row keys |
| `edge_from_id_key_name` | `"from_id"` | Match bulk edge row keys |
| `edge_to_id_key_name` | `"to_id"` | Match bulk edge row keys |
| `label_key_name` | `"type"` | Match stored label key |
| `property_key_name` | `"properties"` | Match stored property container |
| `date_keys` | `{}` | Keep earlier (`min`/`before`) or later (`max`/`after`) timestamps |
| `incremental_keys` | `{}` | Add or subtract numeric properties |
| `validate_avro` | `True` | Avro runs only for labels with a loaded schema |
| `id_whitespace_replacement` | `"_"` | Source-id whitespace only |
| `enable_logging` | `False` | Rotating file log; not a public logger type |
| `log_dir` / `log_file` / `log_level` | see above | Path-like `log_dir` is stored with `os.fspath`. Default `log_level` is INFO. |
| `graph_name` / `graph_description` | `""` | Inspect JSON and Parquet metadata |

### Logging

INFO is operational logging: GraphIngestor initialization, merge summaries, and
bulk ingest start / periodic progress / completion. Successful individual node
and edge creates or updates log at DEBUG with identities only (`uid`,
`from_id` / `to_id`, `type`). They do not write at the default INFO level and
do not include property payloads.

`upsert_bulk_nodes` / `upsert_bulk_edges` log at INFO:

- one start line
- one progress line every `log_every` processed rows (default `100_000`)
- one final summary

Each bulk line includes entity kind (`nodes` or `edges`), label, processed /
created / updated / failed counts, current stored count, and elapsed seconds.
`log_every = None` keeps start and completion and omits intermediate progress.

Do not set `log_level = logging.DEBUG` for multi-million-record ingestion.
That re-enables a file write per successful row. Leave INFO (or higher) for
production bulk loads; use DEBUG only for small targeted investigations.

`date_keys` and `incremental_keys` are stored as given. Unknown methods are
**not** rejected in `__init__`. They raise `ValueError` when an upsert or
merge applies that property (`Unsupported date comparison method` /
`Unsupported increment method`). Valid date methods: `min`, `before`, `max`,
`after`. Valid increment methods: `add`, `subtract`.

Booleans are not treated as numbers for incremental properties.

## 5. Complete ETL workflow

1. Query and normalize in the ETL (not here).
2. Optionally load property maps and Avro schemas.
3. Project rows with `iter_node_records` / `iter_edge_records`, or the eager
   `make_node_list` / `make_edge_list` wrappers, **or** skip those helpers
   when the stream is already standardized (better for large streams).
4. Ingest with `upsert_bulk_nodes` / `upsert_bulk_edges` (or `ingest_*` for
   single records). Create nodes before edges. Bulk ingest performs structural
   validation, canonicalization, and optional Avro checks. Projection helpers
   do **not**.
5. `validate_integrity(raise_on_error = True)` before handoff.
6. `export_parquet(output_dir)` for GraphWarehouse. Use `save(path)` only for
   a trusted pickle.

### Record shapes before and after `make_*_list`

Incoming ETL row (arbitrary keys, not yet GraphIngestor-shaped):

```python
{"person_id": "john@gmail.com", "full_name": "John", "ignored": True}
```

`make_node_list` projects that into a standardized **bulk** mapping. The
identifier is still the source id; it is not yet canonical:

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    validate_avro = False,
    property_mappings = {"email_address": {"full_name": "name"}},
)
projected = ingestor.make_node_list(
    [{"person_id": "john@gmail.com", "full_name": "John", "ignored": True}],
    "email_address",
    uid_key_name = "person_id",
    skip_keys = ("ignored",),
    map_dict = True,
)
assert projected == [{
    "uid": "john@gmail.com",
    "type": "email_address",
    "properties": {"name": "John"},
}]
```

After `upsert_bulk_nodes`, the stored node uses the canonical id:

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    validate_avro = False,
    property_mappings = {"email_address": {"full_name": "name"}},
)
projected = ingestor.make_node_list(
    [{"person_id": "john@gmail.com", "full_name": "John", "ignored": True}],
    "email_address",
    uid_key_name = "person_id",
    skip_keys = ("ignored",),
    map_dict = True,
)
assert ingestor.upsert_bulk_nodes(projected, "email_address")["created"] == 1
stored = ingestor.nodes["email_address:john@gmail.com"]
assert stored["uid"] == "email_address:john@gmail.com"
assert stored["type"] == "email_address"
assert stored["properties"] == {"name": "John"}
```

Incoming edge rows need canonical endpoint ids. After `make_edge_list`:

```python
{
    "from_id": "email_address:john@gmail.com",
    "to_id": "person:Ada_Lovelace",
    "type": "has_owner",
    "properties": {"role": "primary"},
}
```

When the stream is already standardized (`uid` / `from_id` / `to_id` /
`properties`, optional `event_ts`), skip `make_*_list` and pass the mappings
to bulk ingest:

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(weighted = False, directional = True, validate_avro = False)
assert ingestor.upsert_bulk_nodes(
    [{"uid": "john@gmail.com", "properties": {"provider": "gmail"}}],
    "email_address",
)["created"] == 1
assert "email_address:john@gmail.com" in ingestor.nodes
```

`make_node_list` / `make_edge_list` only rename, skip, and nest fields. They
do **not** canonicalize ids, do **not** schema-validate records, and do
**not** run Avro. `upsert_bulk_nodes` / `upsert_bulk_edges` (and the
single-record ingest methods) perform validation, canonicalization, and
optional Avro, including schema-directed coercion when a schema is loaded.

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(weighted = False, directional = True)

node_rows = ingestor.make_node_list(
    [{"person_id": "1", "full_name": "Ada"}],
    "person",
    uid_key_name = "person_id",
)
print(ingestor.upsert_bulk_nodes(node_rows, "person"))
# {"created": 1, "updated": 0, "failed": 0}

ingestor.validate_integrity(raise_on_error = True)
ingestor.export_parquet("warehouse/graph_v1")
```

## 6. Property mappings

`create_property_mapping(directory, suffix = "_mapping.json")` loads JSON
objects. The filename without `suffix` is the label. The result replaces
`property_mappings` atomically and is returned.

Expected file: `person_mapping.json`

```json
{"full_name": "name", "raw_score": "score"}
```

```python
from pathlib import Path
import json
from graph_ingestor import GraphIngestor

mapping_dir = Path("mappings")
mapping_dir.mkdir(exist_ok = True)
(mapping_dir / "person_mapping.json").write_text(
    json.dumps({"full_name": "name"}),
    encoding = "utf-8",
)
(mapping_dir / "knows_mapping.json").write_text(
    json.dumps({"edge_note": "note"}),
    encoding = "utf-8",
)

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    property_mappings = {"legacy": {"old": "new"}},
)
loaded = ingestor.create_property_mapping(mapping_dir)
assert loaded == {
    "knows": {"edge_note": "note"},
    "person": {"full_name": "name"},
}
assert ingestor.property_mappings == loaded
```

An empty directory **clears** existing mappings. Custom `suffix` is supported
(for example `.map.json`). Failures: `FileNotFoundError`,
`NotADirectoryError`, empty suffix, malformed JSON, non-object JSON,
non-string keys/values, duplicate target keys, blank label prefix, duplicate
labels.

`make_node_list` / `make_edge_list` with `map_dict = True` rename keys through
the label's mapping. Extra unmapped source fields are dropped. `skip_keys`
matches source **and** target names. A `source` value overwrites a mapped
property of the same name. This is field projection only; values are not
cleaned.

Node example:

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    property_mappings = {"person": {"full_name": "name", "raw_score": "score"}},
)
rows = ingestor.make_node_list(
    [{"uid": "1", "full_name": "Ada", "raw_score": 3, "ignored": True}],
    "person",
    map_dict = True,
)
assert rows[0]["uid"] == "1"
assert rows[0]["properties"] == {"name": "Ada", "score": 3}
```

Edge example:

```python
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    property_mappings = {"works_at": {"job_title": "role"}},
)
rows = ingestor.make_edge_list(
    [{"from_id": "person:1", "to_id": "org:acme", "job_title": "engineer"}],
    "works_at",
    map_dict = True,
)
assert rows[0]["properties"] == {"role": "engineer"}
```

`map_dict = True` without a mapping for that label raises `KeyError` from
`make_node_list` / `iter_node_records` and `ValueError` from `make_edge_list`
/ `iter_edge_records`.

## 7. Avro mappings and validation

`create_avro_mapping(directory, suffix = "_avro_schema.avsc")` parses Avro
schema files. The filename without `suffix` is the label. Parsed schemas
replace `ingestor.labels` atomically. Loaded Avro schemas must be replaced by
rebinding through `create_avro_mapping(...)` or assignment to
`ingestor.labels`. Mutating a loaded schema dictionary in place is
unsupported.

Property mapping = rename fields. Avro mapping = coerce then type-check
property values with `fastavro`. They are independent.

When `validate_avro = True` (the default) **and** a schema exists for that
label, each record is coerced according to that schema and then passed through
the existing Avro validator:

```
input record → schema-directed coercion → Avro validation → ingest
```

Coercion does not replace validation. Uncoercible or still-invalid values fail
the same way as before: single-record calls return `None` and increment
`failed_node_insertion_count` / `failed_edge_insertion_count`. Bulk methods
count those rows as `failed` and continue. JSON-unsafe and non-finite
property values are rejected at ingest even when Avro is off; they are not an
Avro-only concern. In bulk ingest, expected typed data errors increment
`failed` and continue; unexpected implementation exceptions still abort.

Extra properties, date/increment keys, and `None` values on upserts are not
Avro-checked. A label without a loaded schema still gets structural validation
only. `validate_avro = False` skips both coercion and Avro even if schemas are
loaded. Coercion returns a new mapping and does not mutate the caller's input.

### Primitive conversion table

| Avro type | Accepted inputs | Result | Rejected |
|---|---|---|---|
| `int` | Python `int` (not `bool`); integer strings such as `"12"`, `"-4"`, `"+7"` with surrounding whitespace; integral finite floats such as `12.0` | Python `int` in `[-2**31, 2**31 - 1]` | `bool`; fractional numbers such as `12.5`; `"12.5"`, `"abc"`, `""`; NaN; ±infinity; 32-bit overflow |
| `long` | Same as `int` | Python `int` in `[-2**63, 2**63 - 1]` | Same as `int`, plus 64-bit overflow |
| `float` / `double` | Python `int` and `float` (not `bool`); numeric strings such as `"1"`, `"1.25"`, `"-4.2"` with surrounding whitespace | Python `float` | `bool`; empty or nonnumeric strings; NaN; ±infinity |
| `boolean` | See the boolean table below | Python `bool` | `"yes"`, `"no"`, `2`, `-1`, empty strings, and other truthy/falsy objects. Plain `bool(value)` is not used. |
| `string` | Existing Python `str`; non-null scalars via `str(value)` | Python `str` | `None` is not turned into `"None"`. Dictionaries, lists, and other structured containers are not stringified. |

Boolean inputs:

| Input | Result |
|---|---|
| `True` | `True` |
| `False` | `False` |
| `"true"` | `True` |
| `"false"` | `False` |
| `"1"` | `True` |
| `"0"` | `False` |
| integer `1` | `True` |
| integer `0` | `False` |

String matching is case-insensitive and ignores surrounding whitespace, so
`" TRUE "` is valid.

### Nulls, unions, and nested values

Nullable unions such as `["null", "string"]` and `["null", "long"]` preserve
`None`. Non-nullable `None` is left unchanged and fails Avro validation.

Non-null union values try branches in schema order, skipping `"null"`. The
first branch that both coerces **and** validates wins. `"123"` against
`["int", "string"]` becomes `123`. Against `["string", "int"]` it stays
`"123"`.

Records, nested records, arrays, and maps are coerced recursively. Array
items use the item schema; map values use the value schema. Bytes, enums,
fixed types, logical types, aliases, defaults, extra fields, and named schema
references keep their previous validation behavior.

Valid schema files (from the test suite):

`person_avro_schema.avsc`

```json
{
  "type": "record",
  "name": "Person",
  "fields": [{"name": "name", "type": "string"}]
}
```

`knows_avro_schema.avsc`

```json
{
  "type": "record",
  "name": "Knows",
  "fields": [{"name": "note", "type": "string"}]
}
```

```python
from pathlib import Path
import json
from graph_ingestor import GraphIngestor

schema_dir = Path("schemas")
schema_dir.mkdir(exist_ok = True)
(schema_dir / "person_avro_schema.avsc").write_text(
    json.dumps({
        "type": "record",
        "name": "Person",
        "fields": [
            {"name": "name", "type": "string"},
            {"name": "age", "type": ["null", "int"], "default": None},
        ],
    }),
    encoding = "utf-8",
)

ingestor = GraphIngestor(weighted = False, directional = True, validate_avro = True)
ingestor.create_avro_mapping(schema_dir)

ok = ingestor.create_node_object("person", 1, {"name": "Ada", "age": "30"})
coerced_name = ingestor.create_node_object("person", 2, {"name": 123})
bad = ingestor.create_node_object("person", 3, {"name": "Ada", "age": "twelve"})
assert ok is not None
assert ok["properties"]["age"] == 30
assert coerced_name is not None
assert coerced_name["properties"]["name"] == "123"
assert bad is None
assert ingestor.failed_node_insertion_count == 1
assert ingestor.node_count == 2
```

Avro still does **not** trim stored strings, fill defaults, or normalize case
beyond the boolean table above. Empty schema directories clear `labels`, same
as property mappings.

### Diagnostic validator isolation

`scripts/avro_validator_isolation.py` is a diagnostic-only harness. It runs
GraphIngestor’s current Avro pipeline unchanged except for injecting a
validator into **both** `graph_ingestor.core.avro_validate` and
`graph_ingestor.avro_coerce.avro_validate`. Production package files continue
to import `fastavro.validation.validate` only.

`--mode python` uses `fastavro._validation_py` from the installed wheel. That
module is **not** a supported production dependency and must not be imported
from `graph_ingestor`. Run each `--mode` (`off`, `compiled`, `python`) in a
fresh process. Do not add million-row isolation runs to the normal test suite.

## 8. Node operations

**create vs upsert vs ingest vs bulk**

| Call | Identity input | Missing target | Duplicate | Typical use |
|---|---|---|---|---|
| `create_node_object` | source id, canonicalized | inserts | `ValueError` | first insert |
| `upsert_node_object` | **canonical** id | `KeyError` | updates | known existing node |
| `ingest_node` | source id, canonicalized | create or upsert | updates | preferred single-record |
| `upsert_bulk_nodes` | opaque source id by default; `ids_are_canonical=True` for prefixed uids | create or upsert | updates | ETL streams |

```python
from graph_ingestor import GraphIngestor

g = GraphIngestor(weighted = False, directional = True, validate_avro = False)
created = g.create_node_object("person", 1, {"name": "Ada"})
assert created == {
    "uid": "person:1",
    "type": "person",
    "properties": {"name": "Ada"},
}

updated = g.upsert_node_object("person:1", {"name": "Ada Lovelace"})
assert updated["properties"]["name"] == "Ada Lovelace"

g.ingest_node("person", 2, {"name": "Grace"})
counts = g.upsert_bulk_nodes(
    [{"uid": "3", "properties": {"name": "Alan"}}],
    "person",
)
assert counts == {"created": 1, "updated": 0, "failed": 0}
```

On a normally completed bulk job, `created + updated + failed` equals the
number of input records consumed. Expected typed record-data failures
increment `failed` and later rows still run. A failed update leaves the
prior stored record unchanged: prepare builds a replacement, and commit is a
single assignment. Unexpected implementation exceptions abort; they are not
converted to `failed`. Rejection logs are one bounded WARNING summary
(reason categories and row positions, no raw ids or property payloads) plus
the INFO job summary.

`iter_node_records` lazily projects already-clean rows (or id strings) into
standardized mappings for bulk ingest. Configuration, label, and mapping
checks run when the iterator is created. The source is pulled only as records
are consumed. Yielded records are ordinary caller-owned dictionaries;
`list(iterator)` is safe. Ingestion still copies stored properties, so this
is not a zero-copy path.

`make_node_list` is `list(self.iter_node_records(...))` and **materializes a
list**. String items become ids with empty properties. Identity and root keys
are not copied into `properties`. Projection does **not** canonicalize ids,
does **not** Avro-validate, and does **not** insert anything. Large pipelines
should stream `iter_node_records(...)` (or already-standardized rows) into
`upsert_bulk_nodes`, which performs validation, canonicalization, and
optional Avro.

`iter_nodes(label_name = None)` yields read-only `MappingProxyType` views
without copying the collection. Do not mutate the graph during iteration.

`remove_node(canonical_id)` returns `{"nodes_removed": 0 or 1, "edges_removed": n}`.
Missing ids are a no-op. Empty id raises `ValueError`. Incident edges are
found by scanning the full edge dict (no adjacency index).

`purge_nodes(canonical_ids)` returns the same count shape, aggregated.
`purge_super_nodes(source_ids, label)` canonicalizes each source id with
`label` first. Duplicates do not inflate counts.

## 9. Edge operations

Endpoints must already exist. Otherwise create/ingest raises `ValueError`.

| Call | Identity input | Missing target | Duplicate | Typical use |
|---|---|---|---|---|
| `create_edge_object` | `(from_id, label, to_id)` | `ValueError` if endpoints missing | `ValueError` | first insert |
| `upsert_edge_object` | same tuple | `KeyError` | updates | known existing edge |
| `ingest_edge` | label + two canonical node ids | create or upsert | updates | preferred single-record |
| `upsert_bulk_edges` | configured from/to keys | create or upsert | updates | ETL streams |

```python
from graph_ingestor import GraphIngestor

g = GraphIngestor(weighted = False, directional = True, validate_avro = False)
g.create_node_object("person", 1, {"name": "Ada"})
g.create_node_object("person", 2, {"name": "Grace"})
edge = g.create_edge_object(
    ("person:1", "knows", "person:2"),
    {"note": "colleagues"},
)
assert edge["from_id"] == "person:1"
assert ("person:1", "knows", "person:2") in g.edges

g.ingest_edge("knows", "person:2", "person:1", {"note": "also"})
```

On an undirected graph, `remove_edge(("person:2", "knows", "person:1"))`
normalizes orientation and can remove `("person:1", "knows", "person:2")`.

`iter_edge_records` is the lazy edge counterpart of `iter_node_records`.
`make_edge_list` is `list(self.iter_edge_records(...))` and matches
`make_node_list` for skip/source/`map_dict`.
`from_id_key_name` and `to_id_key_name` are keyword-only.
`uid_key_name` is a deprecated alias for `from_id_key_name`; setting both
raises `ValueError`. Like the node helpers, these do not Avro-validate;
`upsert_bulk_edges` does.

`iter_edges(label_name = None)` is the edge counterpart of `iter_nodes`.
`remove_edge` returns whether the edge existed.
`purge_edges` returns the number actually removed (duplicates/missing ignored).

## 10. Inspection and validation

```python
from graph_ingestor import GraphIngestor

g = GraphIngestor(weighted = False, directional = True, validate_avro = False)
g.create_node_object("person", 1, {"name": "Ada"})
g.create_node_object("person", 2, {"name": "Grace"})
g.create_edge_object(("person:1", "knows", "person:2"), {})

assert g.get_node_list_size() > 0
assert g.get_edge_list_size() > 0
assert set(g.estimate_total_size()) == {"nodes", "edges", "total"}
assert g.count_graph_dimensions() == {"nodes": 2, "edges": 1}
assert g.validate_integrity() == []
g.validate_integrity(raise_on_error = True)
```

Size methods use `sys.getsizeof` recursively. They are interpreter estimates,
not RSS. `total` uses one shared-object tracker and need not equal
nodes + edges. `estimate_total_size` excludes derived Avro cache contents.
`count_graph_dimensions` reads counters; it does not scan or
repair.

`validate_integrity` walks every node and edge: canonical keys, root fields,
endpoint existence, counters, type sets. O(nodes + edges).

Users need to know this public failure split (implemented by private
validators):

- expected per-record **typed data errors** in bulk (`missing` keys, opaque
  identity failures, canonical-prefix mismatch, JSON-unsafe values, Avro
  coerce/validation/`OverflowError` at that boundary, missing edge
  endpoints, edge label mismatch, merge rule failures) → `failed` count,
  processing continues
- unexpected implementation exceptions (`RuntimeError`, stray `TypeError` /
  `ValueError` / `KeyError` from bugs, `RecursionError`) → abort; they are
  not converted to `failed`
- single-record Avro type mismatch → `None` plus the failed insertion
  counter
- wrong Python type or missing target on single-record APIs → `TypeError` /
  `ValueError` / `KeyError` as before

## 11. Persistence and interchange

### Pickle (`save` / `load`)

Writes a gzip pickle to **exactly** `filepath` (no extra suffix). Atomic
replace inside the parent directory. Parent directories are **not** created.

Pickle can execute code on load. Use it only between trusted parties. Parquet
is the interchange format. `GraphIngestor.load()` remains a trusted-input
library API. CLI commands that would unpickle a file (`inspect`, `validate`,
`export-parquet`, `merge`) require `--allow-pickle` and print a trusted-
input / arbitrary-code warning. Without the flag, files are refused before
`pickle.load`; Parquet directories still load normally. Gzip or pickle magic
bytes are not treated as authorization.

The envelope is a dict (`format`, `version`, `object_type`, `state`), not a
raw `GraphIngestor` instance. Runtime logger objects are omitted; logging
config is kept. `load` restores a disabled or file logger from that config.

### `merge_graph_ingestors`

```python
from graph_ingestor import GraphIngestor

left = GraphIngestor(weighted = False, directional = True, validate_avro = False)
right = GraphIngestor(weighted = False, directional = True, validate_avro = False)
left.create_node_object("person", 1, {"name": "Ada"})
right.create_node_object("person", 2, {"name": "Grace"})
assert left.merge_graph_ingestors(right, merge_failed_counts = True) is left
assert left.node_count == 2
```

Merges `other` into `destination` in place and returns `self`. Nodes merge
before edges. Ordinary non-null properties replace; date/increment rules
apply; non-null `event_ts` replaces. Destination `graph_id` and description
are kept. Configurations that must match: `weighted`, `directional`, root key
names, `date_keys`, `incremental_keys`, and Avro `labels`. Merging an object
with itself is a no-op.

### Parquet (`export_parquet` / `load_parquet`)

```python
from pathlib import Path
import tempfile
from graph_ingestor import GraphIngestor

ingestor = GraphIngestor(weighted = False, directional = True, validate_avro = False)
ingestor.create_node_object("person", 1, {"name": "Ada"})
export_dir = Path(tempfile.mkdtemp()) / "graph_v1"
ingestor.export_parquet(export_dir, chunk_size = 250000, overwrite = False)
loaded = GraphIngestor.load_parquet(export_dir, batch_size = 100000)
assert loaded.node_count == 1
```

Layout:

- `metadata.json` — format/version, graph identity, flags, in-memory key
  names, date/increment rules, mappings, counts, manifest
- `nodes/<label>.parquet` — `uid`, `type`, `properties`, `event_ts`
- `edges/<label>.parquet` — `from_id`, `to_id`, `type`, `properties`, `event_ts`

Custom in-memory key names are recorded in metadata. Parquet columns stay
warehouse-compatible. Properties may load from Arrow STRUCT/MAP or JSON
object text. Logging paths are never restored; the loaded instance has
logging disabled.

Existing **managed** files raise `FileExistsError` unless `overwrite = True`.
`overwrite = False` is unchanged: the destination is not replaced.

Overwrite writes the complete new export to a sibling staging directory on
the **same filesystem** (`{name}.export-tmp-*`), validates `metadata.json`,
manifest containment, and Parquet footer row counts (no second in-memory
graph), then journals `{name}.export-journal-*` and swaps: destination →
`{name}.export-bak-*`, staging → destination. If the publication rename
fails, the backup is renamed back. True unmanaged files (for example
`notes.txt`) are copied back from the backup. Obsolete managed Parquet
that the new graph dropped is **not** restored. Concurrent exports to the
same destination are refused while `{name}.export-lock` is held (exclusive
file-descriptor lock; `fcntl` on POSIX, `msvcrt` on Windows). Parent-
directory `fsync` after journal and rename is best-effort and is skipped
where the OS does not support it. This is not an atomic replacement of a
non-empty destination directory; POSIX `rename` onto an occupied non-empty
directory is not portable. A valid previous export remains available until
a complete replacement is ready.

Leftovers after a crash or failed cleanup are siblings of the destination:

- `{name}.export-tmp-*` — incomplete staging; destination is still the
  previous complete export, or absent
- `{name}.export-bak-*` — backup of the previous destination
- `{name}.export-journal-*` — crash journal used by the next
  `export_parquet` to the same path
- `{name}.export-lock` — lock file; may remain after a successful export

Destination and backup together without a coherent journal are refused as
ambiguous: keep the destination or restore the backup by hand, then remove
the leftover backup/journal files. The next successful `export_parquet`
after lock acquire runs that recovery before writing.

Invalid metadata or records abort `load_parquet`; no partial graph is
returned.

This is the GraphWarehouse interchange contract. Compatible Warehouse versions
treat Parquet v1 `metadata.json` as **authoritative**: `format` is
`local-graph-ingestor-parquet`, `version` is the integer `1`, and only
manifest-listed `nodes/<file>.parquet` / `edges/<file>.parquet` files are
ingested. GraphIngestor writes offset-aware UTC ISO-8601 `event_ts` strings;
those Warehouse versions store them as naive UTC DuckDB `TIMESTAMP`. This
README does not describe GraphWarehouse APIs.

## 12. CLI

```bash
graph-ingestor inspect path/to/graph.pkl --allow-pickle
graph-ingestor inspect path/to/parquet_dir
graph-ingestor validate path/to/graph.pkl --allow-pickle
graph-ingestor export-parquet path/to/graph.pkl path/to/parquet_dir --allow-pickle
graph-ingestor export-parquet path/to/graph.pkl path/to/parquet_dir --chunk-size 1 --overwrite --allow-pickle
graph-ingestor load-parquet --directory path/to/parquet_dir --save path/to/graph.pkl
graph-ingestor merge --into dest.pkl --source other.pkl --allow-pickle
```

`python -m graph_ingestor` is equivalent.

`inspect` prints JSON with `graph_id`, `graph_name`, `nodes`, `edges`,
`node_types`, and `edge_types`. Directories are loaded with `load_parquet`.
Files are pickle snapshots and require `--allow-pickle`.

`validate` prints `ok` or one integrity violation per line. A corrupt pickle
prints the `ValueError` and exits `1`.

Exit codes from `main()`:

| Situation | Exit |
|---|---|
| Success (`inspect`, export, load-parquet, merge) | 0 |
| Integrity violations, invalid pickle on `validate`, or pickle file without `--allow-pickle` | 1 |
| Missing file (`FileNotFoundError`) | 2 |

`--overwrite` is required when managed export files already exist. Without
it, `export_parquet` raises `FileExistsError`. The CLI does not catch that
exception today, so it is **not** a stable CLI exit-code contract. Treat it
as a known limitation: pass `--overwrite`, or export to a new directory.

`merge` file operands require `--allow-pickle`. Parquet directories load
without that flag.

## 13. Performance and memory guidance

The completed graph is always in RAM, including after batched `load_parquet`.

- Prefer `upsert_bulk_*` with a generator of standardized rows.
- `iter_node_records` / `iter_edge_records` project without building the full
  list. Yielded dicts are caller-owned; ingestion still copies stored
  properties (not zero-copy).
- `make_node_list` / `make_edge_list` build a full Python list; skip them on
  large streams.
- `chunk_size` (export) and `batch_size` (import) bound Arrow batches, not
  peak graph size.
- Full-graph work: `validate_integrity`, size estimates, `remove_node` /
  `purge_*` incident-edge scans, materializing iterators.
- There is no adjacency index. Removing one node scans every edge.
- For millions of nodes and edges, budget Python object overhead (dicts,
  tuples, strings), not the Parquet file size. Export, then let GraphWarehouse
  own the durable copy. Do not keep a giant `GraphIngestor` as the warehouse.

## 14. Error-handling examples and common mistakes

```python
from datetime import datetime
from graph_ingestor import GraphIngestor

g = GraphIngestor(weighted = False, directional = True, validate_avro = False)
g.create_node_object("person", 1, {"name": "Ada"})

try:
    g.create_node_object("person", 1, {"name": "Ada"})
except ValueError:
    pass  # duplicate create

try:
    g.upsert_node_object("person:999", {"name": "Missing"})
except KeyError:
    pass  # upsert requires an existing canonical id

try:
    g.create_edge_object(("person:1", "knows", "person:2"), {})
except ValueError:
    pass  # endpoint person:2 does not exist yet

try:
    g.create_node_object("person", 3, {"name": "X"}, event_ts = datetime(2026, 1, 1))
except ValueError:
    pass  # naive datetime; include a timezone

try:
    g.create_node_object("person", 4, {"name": "X", "event_ts": "nope"})
except ValueError:
    pass  # event_ts is reserved; cannot live in properties
```

Common mistakes:

- Passing source id `"1"` to `upsert_node_object` instead of `"person:1"`
- Inserting edges before both endpoints exist
- Storing `event_ts` inside `properties`
- Expecting `weighted = True` to add weights
- Expecting Avro to fill defaults or stringify dictionaries
- Mutating `ingestor.nodes` / `ingestor.edges` directly
- Expecting `save()` to create missing parent directories
- Using pickle as interchange instead of Parquet
- `map_dict = True` without a mapping for that label (`KeyError` on nodes,
  `ValueError` on edges)
- Re-exporting to the same Parquet directory without `--overwrite`

## 15. Public-method reference

Internal helpers (`_*`), `LoggerHelper`, and CLI module functions are not
part of this table.

| Method | Purpose | Typical use | Mutates? | Return |
|---|---|---|---|---|
| `GraphIngestor(...)` | Configure an empty graph | Start of an ETL | yes (new instance) | `None` |
| `create_property_mapping` | Load JSON field maps by label | Rename ETL columns | yes | `dict[str, dict[str, str]]` |
| `create_avro_mapping` | Load Avro schemas by label | Property type checks | yes | `dict` of parsed schemas |
| `make_node_list` | Project rows to bulk node mappings | Small/medium ETL | no | `list[dict]` |
| `make_edge_list` | Project rows to bulk edge mappings | Small/medium ETL | no | `list[dict]` |
| `iter_node_records` | Lazily project rows to bulk node mappings | Large ETL streams | no | iterator of `dict` |
| `iter_edge_records` | Lazily project rows to bulk edge mappings | Large ETL streams | no | iterator of `dict` |
| `create_node_object` | Insert one new node | First insert | yes | node dict or `None` |
| `upsert_node_object` | Update one existing node | Known canonical id | yes | node dict or `None` |
| `ingest_node` | Create or upsert one node | Preferred single node | yes | node dict or `None` |
| `upsert_bulk_nodes` | Create or upsert many nodes | Primary node ingest | yes | `{created, updated, failed}`; `log_every` INFO cadence |
| `create_edge_object` | Insert one new edge | First insert | yes | edge dict or `None` |
| `upsert_edge_object` | Update one existing edge | Known edge tuple | yes | edge dict or `None` |
| `ingest_edge` | Create or upsert one edge | Preferred single edge | yes | edge dict or `None` |
| `upsert_bulk_edges` | Create or upsert many edges | Primary edge ingest | yes | `{created, updated, failed}`; `log_every` INFO cadence |
| `iter_nodes` | Yield read-only node views | Inspection | no | iterator |
| `iter_edges` | Yield read-only edge views | Inspection | no | iterator |
| `remove_node` | Delete canonical node + incident edges | Cleanup | yes | `{nodes_removed, edges_removed}` |
| `purge_nodes` | Delete many canonical nodes | Cleanup | yes | `{nodes_removed, edges_removed}` |
| `purge_super_nodes` | Delete by source ids + label | Super-node cleanup | yes | `{nodes_removed, edges_removed}` |
| `remove_edge` | Delete one edge | Cleanup | yes | `bool` |
| `purge_edges` | Delete many edges | Cleanup | yes | `int` |
| `get_node_list_size` | Estimate nodes-dict size | Diagnostics | no | `float` MiB |
| `get_edge_list_size` | Estimate edges-dict size | Diagnostics | no | `float` MiB |
| `estimate_total_size` | Estimate instance size | Diagnostics | no | `{nodes, edges, total}` MiB |
| `count_graph_dimensions` | Report maintained counts | Diagnostics | no | `{nodes, edges}` |
| `validate_integrity` | Full invariant check | Before warehouse handoff | no | `list[str]` |
| `save` | Trusted gzip pickle | Intermediate snapshot | no (filesystem) | `Path` |
| `load` | Load trusted pickle | Resume snapshot | new instance | `GraphIngestor` |
| `export_parquet` | GraphWarehouse export | ETL handoff | filesystem | `Path` |
| `load_parquet` | Load Parquet export | Round-trip / merge prep | new instance | `GraphIngestor` |
| `merge_graph_ingestors` | Merge another graph in place | Combine snapshots | yes | `self` |

## 16. End-to-end example

Multiple labels, property maps, Avro on `person`, bulk ingest, integrity
check, Parquet export, and a trusted pickle. Copy into a working directory
with the package installed.

```python
from datetime import datetime, timezone
from pathlib import Path
import json
import tempfile

from graph_ingestor import GraphIngestor

root = Path(tempfile.mkdtemp(prefix = "graph-ingestor-etl-"))
mapping_dir = root / "mappings"
schema_dir = root / "schemas"
export_dir = root / "warehouse"
pickle_path = root / "graph.pkl"
mapping_dir.mkdir()
schema_dir.mkdir()

(mapping_dir / "person_mapping.json").write_text(
    json.dumps({"full_name": "name"}),
    encoding = "utf-8",
)
(mapping_dir / "org_mapping.json").write_text(
    json.dumps({"org_name": "name"}),
    encoding = "utf-8",
)
(schema_dir / "person_avro_schema.avsc").write_text(
    json.dumps({
        "type": "record",
        "name": "Person",
        "fields": [{"name": "name", "type": "string"}],
    }),
    encoding = "utf-8",
)

ingestor = GraphIngestor(
    weighted = False,
    directional = True,
    validate_avro = True,
    date_keys = {"last_seen": "max"},
    graph_name = "People and orgs",
)
ingestor.create_property_mapping(mapping_dir)
ingestor.create_avro_mapping(schema_dir)

people = ingestor.make_node_list(
    [
        {"uid": "1", "full_name": "Ada"},
        {"uid": "2", "full_name": "Grace"},
    ],
    "person",
    map_dict = True,
)
orgs = ingestor.make_node_list(
    [{"uid": "acme", "org_name": "Acme"}],
    "org",
    map_dict = True,
)
assert ingestor.upsert_bulk_nodes(people, "person")["created"] == 2
assert ingestor.upsert_bulk_nodes(orgs, "org")["created"] == 1

ingestor.ingest_node(
    "person",
    1,
    {"last_seen": "2026-01-02T00:00:00+00:00"},
    event_ts = datetime(2026, 1, 2, tzinfo = timezone.utc),
)

works = ingestor.make_edge_list(
    [{"from_id": "person:1", "to_id": "org:acme", "role": "engineer"}],
    "works_at",
)
knows = ingestor.make_edge_list(
    [{"from_id": "person:1", "to_id": "person:2", "note": "colleagues"}],
    "knows",
)
assert ingestor.upsert_bulk_edges(works, "works_at")["created"] == 1
assert ingestor.upsert_bulk_edges(knows, "knows")["created"] == 1

assert ingestor.count_graph_dimensions() == {"nodes": 3, "edges": 2}
ingestor.validate_integrity(raise_on_error = True)

ingestor.export_parquet(export_dir)
ingestor.save(pickle_path)
loaded = GraphIngestor.load_parquet(export_dir)
assert loaded.node_count == 3
assert loaded.edge_count == 2
metadata = json.loads((export_dir / "metadata.json").read_text(encoding = "utf-8"))
assert metadata["format"] == "local-graph-ingestor-parquet"
assert metadata["version"] == 1
assert metadata["manifest"]["nodes"]
assert metadata["manifest"]["edges"]
print(root)
```

Expected: three nodes (`person:1`, `person:2`, `org:acme`), two edges,
`warehouse/metadata.json` (Parquet v1, authoritative to compatible
GraphWarehouse), label-partitioned Parquet with offset-aware `event_ts`, and
`graph.pkl`. Compatible Warehouse versions store those timestamps as naive
UTC DuckDB `TIMESTAMP`.
