src.dackar.knowledge_graph.kg_ingest_neo4j_workflow

Attributes

LOGGER
handler
JsonLike

Functions

load_json(path)	Load a JSON file and return its contents as a dict, or None if no path given.
load_json_list(paths)	Load a sequence of JSON files, returning only non-empty dict results.
_looks_like_processed_text_record(obj)	Return True if obj has the minimum required fields of a processed_text_record.
_partition_processed_text_records(records)	Split a sequence of records into valid and malformed processed_text_records.
ingest_workflow_case_to_neo4j(client, schema_paths, *)	Build and ingest a full RCA workflow case graph into Neo4j.
_parse_args()	Parse CLI arguments for the Neo4j ingestion script.
main()	Entry point for the CLI ingestion script.

Module Contents

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.LOGGER

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.handler

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.JsonLike

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.load_json(path)

Load a JSON file and return its contents as a dict, or None if no path given.

Parameters:

path (Optional[Union[str, pathlib.Path]]) – Filesystem path to a .json file, or None / empty string.

Returns:

Parsed JSON object as a dict, or None when path is falsy.

Return type:

JsonLike

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.load_json_list(paths)

Load a sequence of JSON files, returning only non-empty dict results.

Silently skips paths whose file parses to something other than a non-empty dict (e.g. a JSON array or an empty object).

Parameters:

paths (Optional[Sequence[Union[str, pathlib.Path]]]) – Sequence of filesystem paths, or None.

Returns:

List of parsed JSON dicts, one per successfully loaded file.

Return type:

List[Dict[str, Any]]

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow._looks_like_processed_text_record(obj)

Return True if obj has the minimum required fields of a processed_text_record.

Validates that record_id, doc_id, and doc_type are strings and that metadata and provenance are dicts. Does not perform full JSON Schema validation.

Parameters:

obj (Dict[str, Any]) – Candidate dict to inspect.

Returns:

True if the dict structurally resembles a valid processed_text_record.

Return type:

bool

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow._partition_processed_text_records(records)

Split a sequence of records into valid and malformed processed_text_records.

Parameters:

records (Optional[Sequence[Dict[str, Any]]]) – Sequence of candidate record dicts, or None.

Returns:

A two-tuple (good, bad) where good contains records that pass the _looks_like_processed_text_record() check and bad contains those that do not.

Return type:

Tuple[List[Dict[str, Any]], List[Dict[str, Any]]]

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.ingest_workflow_case_to_neo4j(client, schema_paths, *, event=None, kg_context=None, telemetry_summary=None, evidence_bundle=None, causality_candidates=None, rca_card=None, operational_context=None, pm_compliance=None, documents=None, processed_text_records=None, database=None, create_constraints=True)

Build and ingest a full RCA workflow case graph into Neo4j.

Orchestrates schema constraint application, graph construction from all supplied artifacts, and bulk ingestion. Malformed processed_text_record entries are filtered out with a warning before the graph is built.

Parameters:

• client (dackar.knowledge_graph.py2neo.Py2Neo) – Active Py2Neo connection.

• schema_paths (Union[str, pathlib.Path, Iterable[Union[str, pathlib.Path]]]) – One or more paths to TOML schema files.

• event (JsonLike) – Parsed event artifact dict.

• kg_context (JsonLike) – Parsed kg_context artifact dict.

• telemetry_summary (JsonLike) – Parsed telemetry_summary artifact dict.

• evidence_bundle (JsonLike) – Parsed evidence_bundle artifact dict.

• causality_candidates (JsonLike) – Parsed causality_candidates artifact dict.

• rca_card (JsonLike) – Parsed rca_card artifact dict.

• operational_context (JsonLike) – Parsed operational_context artifact dict.

• pm_compliance (JsonLike) – Parsed pm_compliance artifact dict.

• documents (Optional[Sequence[Dict[str, Any]]]) – List of document descriptor dicts.

• processed_text_records (Optional[Sequence[Dict[str, Any]]]) – List of processed_text_record dicts; malformed entries are skipped with a warning.

• database (Optional[str]) – Target Neo4j database name; uses the driver default when None.

• create_constraints (bool) – When True (default), DDL constraints and indexes are applied before ingestion.

Returns:

A two-tuple (node_count, edge_count) reflecting the number of nodes and edges written to the database.

Return type:

Tuple[int, int]

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow._parse_args()

Parse CLI arguments for the Neo4j ingestion script.

Returns:

Populated argparse.Namespace with connection settings and optional paths to each artifact type.

Return type:

argparse.Namespace

src.dackar.knowledge_graph.kg_ingest_neo4j_workflow.main()

Entry point for the CLI ingestion script.

Parses command-line arguments, opens a Neo4j connection, runs ingest_workflow_case_to_neo4j() with the provided artifact paths, and logs a completion summary. The Neo4j driver is always closed on exit.

Return type:

None