graph

graph

Lineage graph data structure for representing entity relationships.

LineageEdge dataclass

LineageEdge(
    dependent_node: LineageNode,
    dependency_node: LineageNode,
    metadata: dict[str, Any] = dict(),
)

Represents a directed relationship between two nodes in the lineage graph.

Edges are derived from dependencies and dependents on LineageNodes. The edge direction is: dependent_node depends on dependency_node.

LineageGraph dataclass

LineageGraph(
    nodes: set[LineageNode] = set(),
    edges: set[LineageEdge] = set(),
    metadata: dict[str, Any] = dict(),
)

Format-agnostic lineage graph data structure.

Manages nodes and edges representing lineage relationships. Graph construction is synchronous - operates on pre-loaded DomoLineage_Link objects.

All complex traversal logic should be handled by .Lineage.get(). This class only handles deduplication of nodes and edges.

add_edge

add_edge(edge: LineageEdge) -> None

Add edge to graph (automatically deduplicates by hash/eq).

Parameters:

Name	Type	Description	Default
edge	LineageEdge	LineageEdge to add	required

Source code in src/crew_dcs/integrations/graphs/graph.py

def add_edge(self, edge: LineageEdge) -> None:
    """Add edge to graph (automatically deduplicates by hash/eq).

    Args:
        edge: LineageEdge to add
    """
    self.edges.add(edge)

add_node

add_node(node: LineageNode) -> None

Add node to graph (automatically deduplicates by hash/eq).

Parameters:

Name	Type	Description	Default
node	LineageNode	LineageNode to add	required

Source code in src/crew_dcs/integrations/graphs/graph.py

def add_node(self, node: LineageNode) -> None:
    """Add node to graph (automatically deduplicates by hash/eq).

    Args:
        node: LineageNode to add
    """
    self.nodes.add(node)

from_lineage_links classmethod

from_lineage_links(
    links: list[DomoLineage_Link],
) -> LineageGraph

Build graph from LineageLink list (synchronous operation).

Constructs nodes and edges from pre-loaded lineage links. All complex traversal logic should be handled by .Lineage.get() before calling this method.

This method only: 1. Creates LineageNode objects from links 2. Creates LineageEdge objects from dependencies/dependents 3. Deduplicates nodes and edges (via set operations)

Parameters:

Name	Type	Description	Default
links	list[DomoLineage_Link]	List of LineageLink with loaded entities and populated dependencies/dependents	required

Returns:

Type	Description
LineageGraph	LineageGraph with deduplicated nodes and edges

Source code in src/crew_dcs/integrations/graphs/graph.py

@classmethod
def from_lineage_links(
    cls,
    links: list[DomoLineage_Link],
) -> LineageGraph:
    """Build graph from LineageLink list (synchronous operation).

    Constructs nodes and edges from pre-loaded lineage links.
    All complex traversal logic should be handled by <entity>.Lineage.get()
    before calling this method.

    This method only:
    1. Creates LineageNode objects from links
    2. Creates LineageEdge objects from dependencies/dependents
    3. Deduplicates nodes and edges (via set operations)

    Args:
        links: List of LineageLink with loaded entities and
               populated dependencies/dependents

    Returns:
        LineageGraph with deduplicated nodes and edges
    """
    graph = cls()

    # Dictionary to track nodes by node_key for edge creation
    node_map: dict[tuple[str, str], LineageNode] = {}

    # First pass: Create all nodes
    for link in links:
        node = LineageNode.from_lineage_link(link)
        graph.add_node(node)
        node_map[node.node_key] = node

    # Second pass: Create edges from dependencies and dependents
    for link in links:
        current_node = node_map.get((str(link.id), link.type))
        if not current_node:
            continue

        # Create edges to dependencies (what this node depends on)
        if hasattr(link, "dependencies") and link.dependencies:
            for dep_link in link.dependencies:
                dependency_node = node_map.get((str(dep_link.id), dep_link.type))
                if dependency_node:
                    rel_type = resolve_relationship(
                        current_node.entity_type,
                        dependency_node.entity_type,
                    )
                    edge = LineageEdge(
                        dependent_node=current_node,
                        dependency_node=dependency_node,
                        metadata={"relationship_type": rel_type},
                    )
                    graph.add_edge(edge)

        # Create edges from dependents (what depends on this node)
        if hasattr(link, "dependents") and link.dependents:
            for dep_link in link.dependents:
                dependent_node = node_map.get((str(dep_link.id), dep_link.type))
                if dependent_node:
                    rel_type = resolve_relationship(
                        dependent_node.entity_type,
                        current_node.entity_type,
                    )
                    edge = LineageEdge(
                        dependent_node=dependent_node,
                        dependency_node=current_node,
                        metadata={"relationship_type": rel_type},
                    )
                    graph.add_edge(edge)

    return graph

LineageNode dataclass

LineageNode(id: str, lineage_link: DomoLineage_Link)

Core node representation for lineage graph operations.

Separates data model (LineageNode) from presentation (MermaidNode). Designed for graph traversal, deduplication, and relationship tracking.

All graph construction is synchronous - nodes are created from pre-loaded LineageLink objects, no API calls occur during node creation.

entity property

entity: DomoEntity_w_Lineage | None

Get entity from lineage link.

entity_type property

entity_type: str

Get entity type from lineage link.

name property

name: str | None

Get name from lineage link.

node_key property

node_key: tuple[str, str]

Unique key for graph operations: (id, entity_type).

Always converts ID to string for consistent key comparison.

from_lineage_link classmethod

from_lineage_link(link: DomoLineage_Link) -> LineageNode

Create node from lineage link.

Parameters:

Name	Type	Description	Default
link	DomoLineage_Link	LineageLink (may or may not have loaded entity)	required

Returns:

Type	Description
LineageNode	LineageNode with available metadata

Source code in src/crew_dcs/integrations/graphs/graph.py

@classmethod
def from_lineage_link(cls, link: DomoLineage_Link) -> LineageNode:
    """Create node from lineage link.

    Args:
        link: LineageLink (may or may not have loaded entity)

    Returns:
        LineageNode with available metadata
    """
    return cls(
        id=link.id,
        lineage_link=link,
    )