src.dackar.knowledge_graph.kg_query_utils

Functions

is_upstream(component_a, component_b, neo4j_client[, ...])

Return True if component_a is upstream of component_b in the graph.

resolve_edge_type(component_a, component_b, neo4j_client)

Classify the relationship path between two components.

Module Contents

src.dackar.knowledge_graph.kg_query_utils.is_upstream(component_a, component_b, neo4j_client, database=None)[source]

Return True if component_a is upstream of component_b in the graph.

Walks up to six containment/connectivity hops (has_part_usage, owns_port_usage, connects_port) from component_a toward component_b over element_usage nodes.

Parameters:

  • component_a (Optional[str]) – Source component id (element_usage.id). A falsy value, or component_a == component_b, yields False.

  • component_b (Optional[str]) – Target component id (element_usage.id).

  • neo4j_client (Optional[Any]) – Object exposing query(cypher, params, db=...); when None the function returns False without querying.

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

Returns:

True if a directed path a -> ... -> b exists, otherwise False. Any query error is swallowed and reported as False.

Return type:

bool

src.dackar.knowledge_graph.kg_query_utils.resolve_edge_type(component_a, component_b, neo4j_client, database=None)[source]

Classify the relationship path between two components.

Inspects the relationship types along the (up to six-hop) path from component_a to component_b and buckets them into a single category.

Parameters:

  • component_a (Optional[str]) – Source component id (element_usage.id). A falsy value yields "mixed".

  • component_b (Optional[str]) – Target component id (element_usage.id).

  • neo4j_client (Optional[Any]) – Object exposing query(cypher, params, db=...); when None the function returns "mixed" without querying.

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

Returns:

"containment" if the path uses only has_part_usage edges, "connectivity" if it uses only owns_port_usage / connects_port edges, otherwise "mixed" (also returned when no path is found or a query error occurs).

Return type:

str