grai_graph
graph

GraphManifest Objects

class GraphManifest()

get_node

def get_node(namespace: str, name: str) -> Optional[NodeTypes]

Arguments:

namespace (str): name (str):

Returns:

Graph Objects

class Graph()

add_nodes_from_manifest

def add_nodes_from_manifest()

add_edges_from_manifest

def add_edges_from_manifest()

get_node_id

@lru_cache
def get_node_id(namespace: str, name: str) -> Optional[int]

Arguments:

namespace (str): name (str):

Returns:

get_node

def get_node(namespace: Optional[str] = None,
             name: Optional[str] = None,
             node_id: Optional[int] = None) -> GraiType

Arguments:

  • namespace Optional[str], optional - (Default value = None)
  • name Optional[str], optional - (Default value = None)
  • node_id Optional[int], optional - (Default value = None)

Returns:

label

def label(namespace: str, name: str) -> str

Arguments:

namespace (str): name (str):

Returns:

id_label

def id_label(node_id: int) -> str

Arguments:

node_id (int):

Returns:

relabeled_graph

def relabeled_graph()

process_items

@multimethod
def process_items(vals: Any, version: Any, type: Any) -> List[GraiType]

Arguments:

vals (Any): version (Any): type (Any):

Returns:

process_dict

@process_items.register
def process_dict(dict_item: Dict, version: str, type: str) -> GraiType

Arguments:

dict_item (Dict): version (str): type (str):

Returns:

process_node

@process_items.register
def process_node(item: GraiType, version: str, type: str) -> GraiType

Arguments:

item (GraiType): version (str): type (str):

Returns:

process_sequence

@process_items.register
def process_sequence(item_iter: Sequence, version: str,
                     type: str) -> List[GraiType]

Arguments:

item_iter (Sequence): version (str): type (str):

Returns:

build_graph

def build_graph(nodes: List[Dict], edges: List[Dict], version: str) -> Graph

Arguments:

nodes (List[Dict]): edges (List[Dict]): version (str):

Returns:

BaseSourceSegment Objects

class BaseSourceSegment()

__init__

def __init__(node_source_map: Dict[UUID, Iterable[str]],
             edge_map: Dict[UUID, Sequence[UUID]])

Attributes:

  • node_source_map - A dictionary mapping between node id's and the set of source labels for the node
  • node_map - A dictionary mapping source node id's to destination node id's

covering_set

@cached_property
def covering_set() -> tuple[CoveringSourceSet]

Returns:

A tuple of all covering sources

node_cover_map

@cached_property
def node_cover_map() -> Dict[UUID, str]

Returns:

A mapping between node id's and their covering source label

cover_edge_map

@cached_property
def cover_edge_map() -> Dict[str, List[str]]

Returns:

A mapping between covering source labels and covering destination labels