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:

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

pip install mini-atlas-graph-ingestor

For local development:

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:

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:

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

{
    "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.

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.

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

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

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

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:

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:

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

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.

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

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

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:

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:

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

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

knows_avro_schema.avsc

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

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

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

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

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

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)

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

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

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.

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.