graphs

graphs

Graph export converters for lineage data.

Provides converters to transform LineageGraph into various export formats (Mermaid, Neo4j).

Usage

from crew_dcs.integrations.graphs import MermaidConverter

diagram = MermaidConverter.from_lineage_links(links).export(direction="LR")

GraphConverter

Bases: ABC

Abstract base converter for lineage graph export.

Converters transform LineageGraph (format-agnostic data model) into specific export formats (Mermaid, Neo4j, GraphML, etc.).

All conversion is synchronous - operates on pre-loaded graph data.

convert abstractmethod

convert(graph: LineageGraph, **options) -> Any

Convert graph to target format.

Parameters:

Name	Type	Description	Default
graph	LineageGraph	LineageGraph with pre-loaded nodes and edges	required
**options		Format-specific conversion options	{}

Returns:

Type	Description
Any	Format-specific output (MermaidDiagram, Cypher queries, etc.)

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

@abstractmethod
def convert(
    self,
    graph: LineageGraph,
    **options,
) -> Any:
    """Convert graph to target format.

    Args:
        graph: LineageGraph with pre-loaded nodes and edges
        **options: Format-specific conversion options

    Returns:
        Format-specific output (MermaidDiagram, Cypher queries, etc.)
    """

export_to_file abstractmethod

export_to_file(
    output: Any, file_path: str, **options
) -> None

Export converted output to file.

Parameters:

Name	Type	Description	Default
output	Any	Result from convert() method	required
file_path	str	Destination file path	required
**options		Format-specific export options	{}

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

@abstractmethod
def export_to_file(
    self,
    output: Any,
    file_path: str,
    **options,
) -> None:
    """Export converted output to file.

    Args:
        output: Result from convert() method
        file_path: Destination file path
        **options: Format-specific export options
    """

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

MermaidConverter

MermaidConverter(graph: LineageGraph | None = None)

Bases: GraphConverter

Convert lineage graph to Mermaid diagram.

Initialize MermaidConverter with optional LineageGraph.

Parameters:

Name	Type	Description	Default
graph	LineageGraph | None	Optional LineageGraph to convert	None

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

def __init__(self, graph: LineageGraph | None = None):
    """Initialize MermaidConverter with optional LineageGraph.

    Args:
        graph: Optional LineageGraph to convert
    """
    self.graph = graph

convert

convert(
    graph: LineageGraph | None = None,
    direction: str = "TD",
    include_entity_ids: bool = True,
    title: str | None = None,
    trace_direction: Literal[
        "trace_dependencies", "trace_dependents"
    ] = "trace_dependencies",
    **options
) -> MermaidDiagram

Convert LineageGraph to MermaidDiagram.

Parameters:

Name	Type	Description	Default
graph	LineageGraph | None	LineageGraph to convert. If None, uses self.graph	None
direction	str	Mermaid flow direction (default: "TD")	'TD'
include_entity_ids	bool	Whether to include entity IDs	True
title	str | None	Title for the diagram	None
trace_direction	Literal['trace_dependencies', 'trace_dependents']	Direction to trace dependencies	'trace_dependencies'
**options		Additional options	{}

Returns:

Type	Description
MermaidDiagram	MermaidDiagram representation of the graph

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

def convert(
    self,
    graph: LineageGraph | None = None,
    direction: str = "TD",
    include_entity_ids: bool = True,
    title: str | None = None,
    trace_direction: Literal[
        "trace_dependencies", "trace_dependents"
    ] = "trace_dependencies",
    **options,
) -> MermaidDiagram:
    """Convert LineageGraph to MermaidDiagram.

    Args:
        graph: LineageGraph to convert. If None, uses self.graph
        direction: Mermaid flow direction (default: "TD")
        include_entity_ids: Whether to include entity IDs
        title: Title for the diagram
        trace_direction: Direction to trace dependencies
        **options: Additional options

    Returns:
        MermaidDiagram representation of the graph
    """
    _ = include_entity_ids
    _ = options

    target_graph = graph or self.graph
    if not target_graph:
        raise ValueError(
            "No graph provided. Pass graph to convert() or use from_lineage_graph()"
        )

    diagram = MermaidDiagram(
        direction=direction,
        title=title,
    )

    mermaid_nodes: dict[tuple[str, str], MermaidNode] = {}

    for node in target_graph.nodes:
        mermaid_node = MermaidNode.from_lineage_node(node=node)
        diagram.add_node(mermaid_node)
        mermaid_nodes[node.node_key] = mermaid_node

    for edge in target_graph.edges:
        dependent = mermaid_nodes.get(edge.dependent_node.node_key)
        dependency = mermaid_nodes.get(edge.dependency_node.node_key)

        if dependent and dependency:
            rel_type = edge.metadata.get("relationship_type", "DEPENDS_ON")
            if trace_direction == "trace_dependencies":
                relationship = MermaidRelationship(
                    from_node=dependent,
                    to_node=dependency,
                    type=rel_type,
                )
            else:
                relationship = MermaidRelationship(
                    from_node=dependency,
                    to_node=dependent,
                    type=rel_type,
                )
            diagram.add_relationship(relationship)

    return diagram

export

export(**options) -> MermaidDiagram

Convert the stored lineage graph to a MermaidDiagram.

Parameters:

Name	Type	Description	Default
**options		Options to pass to convert()	{}

Returns:

Type	Description
MermaidDiagram	MermaidDiagram representation of the graph

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

def export(self, **options) -> MermaidDiagram:
    """Convert the stored lineage graph to a MermaidDiagram.

    Args:
        **options: Options to pass to convert()

    Returns:
        MermaidDiagram representation of the graph
    """
    return self.convert(**options)

export_to_file

export_to_file(
    file_path: str,
    output: MermaidDiagram | None = None,
    footer_text: str | None = None,
    **options
) -> None

Export Mermaid diagram to markdown file.

Parameters:

Name	Type	Description	Default
file_path	str	Path to output markdown file	required
output	MermaidDiagram | None	MermaidDiagram to export. If None, converts self.graph	None
footer_text	str | None	Optional footer text for the markdown	None
**options		Additional options to pass to convert()	{}

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

def export_to_file(
    self,
    file_path: str,
    output: MermaidDiagram | None = None,
    footer_text: str | None = None,
    **options,
) -> None:
    """Export Mermaid diagram to markdown file.

    Args:
        file_path: Path to output markdown file
        output: MermaidDiagram to export. If None, converts self.graph
        footer_text: Optional footer text for the markdown
        **options: Additional options to pass to convert()
    """
    if output is None:
        output = self.convert(**options)

    output.export_to_markdown(
        export_file=file_path,
        footer_text=footer_text,
    )

from_lineage_graph classmethod

from_lineage_graph(graph: LineageGraph) -> MermaidConverter

Create a MermaidConverter from an existing LineageGraph.

Parameters:

Name	Type	Description	Default
graph	LineageGraph	Existing LineageGraph	required

Returns:

Type	Description
MermaidConverter	MermaidConverter wrapping the graph

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

@classmethod
def from_lineage_graph(cls, graph: LineageGraph) -> MermaidConverter:
    """Create a MermaidConverter from an existing LineageGraph.

    Args:
        graph: Existing LineageGraph

    Returns:
        MermaidConverter wrapping the graph
    """
    return cls(graph=graph)

from_lineage_links classmethod

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

Create a MermaidConverter from lineage links.

Parameters:

Name	Type	Description	Default
links	list[DomoLineage_Link]	List of DomoLineage_Link objects	required

Returns:

Type	Description
MermaidConverter	MermaidConverter with graph created from links

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_converter.py

@classmethod
def from_lineage_links(
    cls,
    links: list[DomoLineage_Link],
) -> MermaidConverter:
    """Create a MermaidConverter from lineage links.

    Args:
        links: List of DomoLineage_Link objects

    Returns:
        MermaidConverter with graph created from links
    """
    graph = LineageGraph.from_lineage_links(links)
    return cls(graph=graph)

MermaidDiagram dataclass

MermaidDiagram(
    nodes: list[MermaidNode] = list(),
    relationships: list[MermaidRelationship] = list(),
    direction: str = "TD",
    title: str | None = None,
    entity_id: str | None = None,
    entity_type: str | None = None,
    use_aliases: bool = True,
)

Complete Mermaid diagram with nodes and relationships.

add_entity_dependency

add_entity_dependency(
    from_entity: Any,
    to_entity: Any,
    *,
    relationship_type: str = "dependency"
) -> None

Add a relationship between two Domo entities.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_entity_dependency(
    self,
    from_entity: Any,
    to_entity: Any,
    *,
    relationship_type: str = "dependency",
) -> None:
    """Add a relationship between two Domo entities."""
    if not from_entity or not to_entity:
        return

    from_node = MermaidNode.from_domo_entity(from_entity)
    to_node = MermaidNode.from_domo_entity(to_entity)
    self.add_node(from_node)
    self.add_node(to_node)

    from_ref = self.find_node(from_entity.id, from_entity.entity_type)
    to_ref = self.find_node(to_entity.id, to_entity.entity_type)

    if from_ref and to_ref and from_ref != to_ref:
        self.add_relationship(
            MermaidRelationship(
                from_node=from_ref,
                to_node=to_ref,
                type=relationship_type,
            )
        )

add_federation_node

add_federation_node(
    entity: Any,
    *,
    custom_type: str | None = None,
    show_entity_id: bool = False,
    show_federation_id: str | None = None
) -> MermaidNode

Add a node with federation-aware type label.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_federation_node(
    self,
    entity: Any,
    *,
    custom_type: str | None = None,
    show_entity_id: bool = False,
    show_federation_id: str | None = None,
) -> MermaidNode:
    """Add a node with federation-aware type label."""
    label_parts = []

    name_line = entity.entity_name
    if hasattr(entity, "auth") and entity.auth:
        name_line += f" [{entity.auth.domo_instance}]"
    label_parts.append(name_line)

    if show_entity_id:
        label_parts.append(str(entity.id))

    if show_federation_id:
        label_parts.append(str(show_federation_id))

    label = "<br/>".join(label_parts)

    if custom_type:
        node_type = custom_type
    elif hasattr(entity, "is_federated") and entity.is_federated:
        node_type = f"Federated{entity.entity_type.title()}"
    else:
        if entity.entity_type == "DATA_SOURCE" and hasattr(
            entity, "data_provider_type"
        ):
            provider_type = entity.data_provider_type or None
            if (
                not provider_type
                and hasattr(entity, "stream_id")
                and not entity.stream_id
            ):
                provider_type = "domo"
            node_type = (
                provider_type.replace("-", " ").title()
                if provider_type
                else "Dataset"
            )
        else:
            node_type = entity.entity_type

    node = MermaidNode(
        id=entity.id,
        name=label,
        type=node_type,
    )
    self.add_node(node)
    return node

add_lineage_items

add_lineage_items(
    lineage_items: Iterable[Any] | None,
    *,
    recursive: bool = False
) -> None

Add a collection of lineage items, optionally following dependencies.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_lineage_items(
    self,
    lineage_items: Iterable[Any] | None,
    *,
    recursive: bool = False,
) -> None:
    """Add a collection of lineage items, optionally following dependencies."""
    if not lineage_items:
        return

    visited: set[tuple[str, str]] = set()

    def _add(link: Any) -> None:
        if not link or not getattr(link, "entity", None):
            return

        entity = link.entity
        key = (str(getattr(entity, "id", "")), getattr(entity, "entity_type", ""))
        if key in visited:
            return
        visited.add(key)

        self.add_lineage_link(link)

        if recursive and getattr(link, "dependencies", None):
            for dependency in link.dependencies:
                _add(dependency)

    for root in lineage_items:
        _add(root)

add_lineage_link

add_lineage_link(lineage_link: Any) -> None

Add a lineage link's node and relationships to the diagram.

Handles both fully-hydrated links (with entity objects) and placeholder links (ID + type only, entity=None). Relationship types are derived from entity types via resolve_relationship (e.g., DATA_SOURCE→DATAFLOW becomes CREATED_BY, PAGE→CARD becomes CONTAINS).

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_lineage_link(self, lineage_link: Any) -> None:
    """Add a lineage link's node and relationships to the diagram.

    Handles both fully-hydrated links (with entity objects) and placeholder
    links (ID + type only, entity=None). Relationship types are derived from
    entity types via resolve_relationship (e.g., DATA_SOURCE→DATAFLOW becomes
    CREATED_BY, PAGE→CARD becomes CONTAINS).
    """
    if not lineage_link:
        return

    source_node = MermaidNode.from_lineage_link(lineage_link)
    self.add_node(source_node)

    # Resolve source_node from the diagram (may have been deduplicated)
    source_type = lineage_link.type
    source_id = str(lineage_link.id)
    source_node = self.find_node(source_id, source_type) or source_node

    if lineage_link.dependencies:
        for dependency in lineage_link.dependencies:
            if not dependency:
                continue

            dep_node = MermaidNode.from_lineage_link(dependency)
            self.add_node(dep_node)

            dep_type = dependency.type
            dep_id = str(dependency.id)
            dep_node = self.find_node(dep_id, dep_type) or dep_node

            if source_node != dep_node:
                rel_type = resolve_relationship(source_node.type, dep_node.type)
                rel = MermaidRelationship(
                    from_node=source_node,
                    to_node=dep_node,
                    type=rel_type,
                )
                self.add_relationship(rel)

add_node

add_node(node: MermaidNode) -> None

Add a node to the diagram if it doesn't already exist.

If a node with the same id+type already exists, the new node's data is merged into the existing one — specifically, a hydrated name replaces a placeholder UUID name, and an explicit shape replaces None.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_node(self, node: MermaidNode) -> None:
    """Add a node to the diagram if it doesn't already exist.

    If a node with the same id+type already exists, the new node's data
    is merged into the existing one — specifically, a hydrated name replaces
    a placeholder UUID name, and an explicit shape replaces None.
    """
    existing = self.find_node(node.id, node.type)
    if existing:
        # Upgrade: hydrated name replaces placeholder UUID
        if node.name != node.id and existing.name == existing.id:
            existing.name = node.name
        # Upgrade: explicit shape replaces None
        if node.shape and not existing.shape:
            existing.shape = node.shape
        # Upgrade: css_class
        if node.css_class and not existing.css_class:
            existing.css_class = node.css_class
        # Upgrade: url
        if node.url and not existing.url:
            existing.url = node.url
        # Upgrade: subtitle
        if node.subtitle and not existing.subtitle:
            existing.subtitle = node.subtitle
        # Upgrade: domo_instance
        if node.domo_instance and not existing.domo_instance:
            existing.domo_instance = node.domo_instance
        return

    if self.use_aliases and node.alias is None:
        node.alias = self._generate_alias(len(self.nodes))
    self.nodes.append(node)

add_relationship

add_relationship(relationship: MermaidRelationship) -> None

Add a relationship to the diagram if it doesn't already exist.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def add_relationship(self, relationship: MermaidRelationship) -> None:
    """Add a relationship to the diagram if it doesn't already exist."""
    if relationship not in self.relationships:
        self.relationships.append(relationship)

combine classmethod

combine(
    diagrams: list[MermaidDiagram],
    *,
    bridge_start_id: str,
    bridge_end_id: str,
    bridge_type: str = "published",
    direction: str | None = None,
    use_aliases: bool | None = None
) -> MermaidDiagram

Merge multiple diagrams and add a bridge edge between two nodes.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

@classmethod
def combine(
    cls,
    diagrams: list[MermaidDiagram],
    *,
    bridge_start_id: str,
    bridge_end_id: str,
    bridge_type: str = "published",
    direction: str | None = None,
    use_aliases: bool | None = None,
) -> MermaidDiagram:
    """Merge multiple diagrams and add a bridge edge between two nodes."""
    if not diagrams:
        raise ValueError("No diagrams provided to combine.")

    combined = cls(
        nodes=[],
        relationships=[],
        direction=direction or diagrams[0].direction,
        use_aliases=(
            use_aliases if use_aliases is not None else diagrams[0].use_aliases
        ),
    )

    for diagram in diagrams:
        for node in diagram.nodes:
            combined.add_node(node)
        for relationship in diagram.relationships:
            combined.add_relationship(relationship)

    if combined.use_aliases:
        for idx, node in enumerate(combined.nodes):
            node.alias = combined._generate_alias(idx)

    def _find_node_by_id(node_id: str) -> MermaidNode | None:
        return next((n for n in combined.nodes if str(n.id) == str(node_id)), None)

    start_node = _find_node_by_id(bridge_start_id)
    end_node = _find_node_by_id(bridge_end_id)

    if not start_node or not end_node:
        raise ValueError(
            "Bridge endpoints not found in combined diagram: "
            f"start={bridge_start_id}, end={bridge_end_id}"
        )

    combined.add_relationship(
        MermaidRelationship(
            from_node=start_node,
            to_node=end_node,
            type=bridge_type,
        )
    )

    return combined

export_to_markdown

export_to_markdown(
    export_file: str | None = None,
    footer_text: str | None = None,
    *,
    show_instance: bool = False,
    show_entity_id: bool = False
) -> str

Render the diagram as a Markdown document with optional file export.

Parameters:

Name	Type	Description	Default
export_file	str | None	Path to write the markdown file	None
footer_text	str | None	Optional footer text	None
show_instance	bool	If True, include domo_instance on node line 3	False
show_entity_id	bool	If True, include entity ID on node line 4	False

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def export_to_markdown(
    self,
    export_file: str | None = None,
    footer_text: str | None = None,
    *,
    show_instance: bool = False,
    show_entity_id: bool = False,
) -> str:
    """Render the diagram as a Markdown document with optional file export.

    Args:
        export_file: Path to write the markdown file
        footer_text: Optional footer text
        show_instance: If True, include domo_instance on node line 3
        show_entity_id: If True, include entity ID on node line 4
    """
    lines = []

    if self.title:
        lines.append(f"# {self.title}")
        lines.append("")

    if self.entity_id and self.entity_type:
        lines.append(f"**Entity Type**: {self.entity_type}")
        lines.append(f"**Entity ID**: {self.entity_id}")
        lines.append("")

    lines.append(f"**Nodes**: {len(self.nodes)}")
    lines.append(f"**Relationships**: {len(self.relationships)}")
    lines.append("")

    lines.append("```mermaid")
    lines.append(
        self.to_string(
            show_instance=show_instance,
            show_entity_id=show_entity_id,
        )
    )
    lines.append("```")

    if footer_text:
        lines.append("")
        lines.append(footer_text.strip())

    markdown_content = "\n".join(lines)

    if export_file:
        output_path = Path(export_file)
        output_path.write_text(markdown_content, encoding="utf-8")

    return markdown_content

find_node

find_node(
    node_id: str, node_type: str
) -> MermaidNode | None

Find a node by ID and type.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def find_node(self, node_id: str, node_type: str) -> MermaidNode | None:
    """Find a node by ID and type."""
    return next(
        (n for n in self.nodes if n.id == node_id and n.type == node_type), None
    )

from_entity classmethod

from_entity(
    entity: Any,
    direction: str = "TD",
    use_aliases: bool = True,
) -> MermaidDiagram

Create a Mermaid diagram shell for a Domo entity.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

@classmethod
def from_entity(
    cls,
    entity: Any,
    direction: str = "TD",
    use_aliases: bool = True,
) -> MermaidDiagram:
    """Create a Mermaid diagram shell for a Domo entity."""
    entity_display_name = _resolve_entity_name(entity)
    diagram = cls(
        nodes=[],
        direction=direction,
        title=f"Lineage Diagram for {entity_display_name}",
        entity_id=str(entity.id),
        entity_type=entity.entity_type,
        use_aliases=use_aliases,
    )
    root_node = MermaidNode.from_domo_entity(entity)
    diagram.add_node(root_node)
    return diagram

from_lineage classmethod

from_lineage(
    entity: Any,
    lineage: list[Any],
    direction: str = "TD",
    use_aliases: bool = True,
) -> MermaidDiagram

Create a fully populated MermaidDiagram from an entity and its lineage.

Combines from_entity + add_lineage_items into a single call. Relationship types are derived from entity types via resolve_relationship.

Parameters:

Name	Type	Description	Default
entity	Any	The root Domo entity	required
lineage	list[Any]	Pre-fetched lineage items (from entity.Lineage.get())	required
direction	str	Mermaid flow direction	'TD'
use_aliases	bool	Whether to use short aliases for node IDs	True

Returns:

Type	Description
MermaidDiagram	MermaidDiagram with all nodes and relationships populated

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

@classmethod
def from_lineage(
    cls,
    entity: Any,
    lineage: list[Any],
    direction: str = "TD",
    use_aliases: bool = True,
) -> MermaidDiagram:
    """Create a fully populated MermaidDiagram from an entity and its lineage.

    Combines from_entity + add_lineage_items into a single call.
    Relationship types are derived from entity types via resolve_relationship.

    Args:
        entity: The root Domo entity
        lineage: Pre-fetched lineage items (from entity.Lineage.get())
        direction: Mermaid flow direction
        use_aliases: Whether to use short aliases for node IDs

    Returns:
        MermaidDiagram with all nodes and relationships populated
    """
    diagram = cls.from_entity(entity, direction=direction, use_aliases=use_aliases)
    diagram.add_lineage_items(lineage, recursive=True)
    return diagram

print_lineage_summary

print_lineage_summary(lineage: list[Any]) -> None

Print a formatted summary of lineage data.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def print_lineage_summary(self, lineage: list[Any]) -> None:  # noqa: C901
    """Print a formatted summary of lineage data."""
    if self.nodes:
        parent_name = self.nodes[0].name
    elif self.title:
        parent_name = self.title.replace("Lineage Diagram for ", "")
    else:
        parent_name = self.entity_id or "Unknown"

    print(f"\n{'=' * 60}")
    print(f"Lineage Summary: {parent_name}")
    if self.entity_type:
        print(f"Type: {self.entity_type}, ID: {self.entity_id}")
    print(f"{'=' * 60}")
    print(f"\nTotal lineage items: {len(lineage)}")

    if lineage:
        by_type = {}
        for item in lineage:
            item_type = item.type
            if item_type not in by_type:
                by_type[item_type] = []
            by_type[item_type].append(item)

        print("\nBy Type:")
        for item_type, items in sorted(by_type.items()):
            print(f"  {item_type}: {len(items)}")

        print("\nDetailed Items:")
        for idx, item in enumerate(lineage, 1):
            if (
                hasattr(item, "entity")
                and item.entity
                and hasattr(item.entity, "entity_name")
            ):
                entity_name = item.entity.entity_name
            else:
                if hasattr(item, "entity") and item.entity:
                    entity_name = (
                        getattr(item.entity, "name", None)
                        or getattr(item.entity, "title", None)
                        or getattr(item.entity, "display_name", None)
                        or item.id
                    )
                else:
                    entity_name = item.id
            print(f"\n  {idx}. {item.type}: {item.id}")
            print(f"     Name: {entity_name}")
            if hasattr(item, "dependencies") and item.dependencies:
                print(f"     Dependencies: {len(item.dependencies)}")
            if hasattr(item, "dependents") and item.dependents:
                print(f"     Dependents: {len(item.dependents)}")
    else:
        print("No lineage items found.")

    print(f"\n{'=' * 60}\n")

to_string

to_string(
    *,
    show_instance: bool = False,
    show_entity_id: bool = False
) -> str

Render the complete Mermaid diagram.

Parameters:

Name	Type	Description	Default
show_instance	bool	If True, include domo_instance on node line 3	False
show_entity_id	bool	If True, include entity ID on node line 4	False

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_diagram.py

def to_string(
    self,
    *,
    show_instance: bool = False,
    show_entity_id: bool = False,
) -> str:
    """Render the complete Mermaid diagram.

    Args:
        show_instance: If True, include domo_instance on node line 3
        show_entity_id: If True, include entity ID on node line 4
    """
    lines = [f"flowchart {self.direction}"]

    for node in self.nodes:
        lines.append(
            node.to_string(
                use_alias=self.use_aliases,
                show_instance=show_instance,
                show_entity_id=show_entity_id,
            )
        )

    if self.relationships:
        lines.append("")
        for rel in self.relationships:
            lines.append(rel.to_string(use_alias=self.use_aliases))

    classes_added = set()
    for node in self.nodes:
        if node.css_class and node.css_class not in classes_added:
            if self.use_aliases and node.alias:
                node_id = node.alias
            else:
                node_id = f"{node.type}_{node.id}".replace("-", "_").replace(
                    " ", "_"
                )[:50]
            lines.append(f"    class {node_id} {node.css_class}")
            classes_added.add(node.css_class)

    # Click directives — link nodes to their Domo instance URLs
    for node in self.nodes:
        if node.url:
            if self.use_aliases and node.alias:
                node_id = node.alias
            else:
                node_id = f"{node.type}_{node.id}".replace("-", "_").replace(
                    " ", "_"
                )[:50]
            safe_name = node.name.replace('"', "'")
            safe_id = str(node.id).replace('"', "'")
            lines.append(
                f'    click {node_id} "{node.url}" "{safe_name} ({safe_id})"'
            )

    return "\n".join(lines)

MermaidNode dataclass

MermaidNode(
    id: str,
    name: str,
    type: str,
    subtitle: str | None = None,
    domo_instance: str | None = None,
    alias: str | None = None,
    css_class: str | None = None,
    shape: str | None = None,
    url: str | None = None,
)

Represents a node in a Mermaid diagram.

Structured label (rendered at to_string time): Line 1: name (entity title) Line 2: subtitle (chart_type, data_provider_type, or class name) Line 3: domo_instance (if show_instance=True) Line 4: id (if show_entity_id=True)

Attributes:

Name	Type	Description
id	str	Unique identifier for the node
name	str	Display name for the entity (title only — no metadata baked in)
type	str	Entity type (e.g., PAGE, CARD, DATA_SOURCE)
subtitle	str | None	Second line — chart_type, data_provider_type, or class name
domo_instance	str | None	Domo instance name (for federated/multi-instance diagrams)
alias	str | None	Short alias for the node (e.g., "A", "B", "C") - auto-generated
css_class	str | None	Optional CSS class for styling (e.g., "FederatedDataset")
shape	str | None	Mermaid shape name (e.g., "cyl", "notch-rect", "fr-rect")
url	str | None	Optional URL for click directive linking to the entity in Domo

mermaid_id property

mermaid_id: str

Generate a Mermaid-safe node ID.

from_domo_entity classmethod

from_domo_entity(entity: Any) -> MermaidNode

Create a MermaidNode from a Domo entity (DomoPage, DomoDataset, etc.).

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_node.py

@classmethod
def from_domo_entity(cls, entity: Any) -> MermaidNode:
    """Create a MermaidNode from a Domo entity (DomoPage, DomoDataset, etc.)."""
    name = _resolve_entity_name(entity)
    url = getattr(entity, "display_url", None)
    subtitle = _resolve_subtitle(entity)
    domo_instance = _resolve_domo_instance(entity)

    return cls(
        id=str(entity.id),
        name=name,
        type=entity.entity_type,
        subtitle=subtitle,
        domo_instance=domo_instance,
        url=url,
    )

from_lineage_link classmethod

from_lineage_link(lineage_link: Any) -> MermaidNode

Create a MermaidNode from a lineage link item.

Structured label: - Line 1: entity name - Line 2: chart_type / data_provider_type / class name - Line 3: domo_instance (stored but only shown if flag is set)

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_node.py

@classmethod
def from_lineage_link(cls, lineage_link: Any) -> MermaidNode:
    """Create a MermaidNode from a lineage link item.

    Structured label:
    - Line 1: entity name
    - Line 2: chart_type / data_provider_type / class name
    - Line 3: domo_instance (stored but only shown if flag is set)
    """
    if lineage_link.entity:
        entity = lineage_link.entity
        name = _resolve_entity_name(entity)
        url = getattr(entity, "display_url", None)
        subtitle = _resolve_subtitle(entity)
        domo_instance = _resolve_domo_instance(entity)

        return cls(
            id=str(entity.id),
            name=name,
            type=entity.entity_type,
            subtitle=subtitle,
            domo_instance=domo_instance,
            url=url,
        )

    return cls(
        id=str(lineage_link.id),
        name=str(lineage_link.id),
        type=lineage_link.type,
    )

from_lineage_node classmethod

from_lineage_node(node: LineageNode) -> MermaidNode

Create a MermaidNode from a LineageNode with formatted display.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_node.py

@classmethod
def from_lineage_node(
    cls,
    node: LineageNode,
) -> MermaidNode:
    """Create a MermaidNode from a LineageNode with formatted display."""
    entity_name = node.name or str(node.id)

    subtitle = None
    domo_instance = None

    if node.entity:
        domo_instance = _resolve_domo_instance(node.entity)
        subtitle = _resolve_subtitle(node.entity)

    if not subtitle:
        # Fallback: entity_type as title case
        subtitle = node.entity_type.replace("_", " ").title()

    css_class = None
    if node.entity and getattr(node.entity, "is_federated", False):
        clean_type = node.entity_type.replace("_", "").title()
        css_class = f"Federated{clean_type}"
    elif node.entity_type in ("SUBSCRIPTION", "PUBLICATION"):
        css_class = node.entity_type.title()

    return cls(
        id=str(node.id),
        name=entity_name,
        type=node.entity_type,
        subtitle=subtitle,
        domo_instance=domo_instance,
        css_class=css_class,
    )

to_string

to_string(
    use_alias: bool = False,
    indentation: int = 4,
    *,
    show_instance: bool = False,
    show_entity_id: bool = False
) -> str

Render the node as a Mermaid node definition.

Uses the v11.3.0+ @{ shape: <name> } syntax when a shape is set, falling back to the bracket syntax (["name"]) otherwise.

Parameters:

Name	Type	Description	Default
use_alias	bool	If True, use the alias instead of full ID	False
indentation	int	Number of spaces to indent	4
show_instance	bool	If True, include domo_instance on line 3	False
show_entity_id	bool	If True, include entity ID on line 4	False

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_node.py

def to_string(
    self,
    use_alias: bool = False,
    indentation: int = 4,
    *,
    show_instance: bool = False,
    show_entity_id: bool = False,
) -> str:
    """Render the node as a Mermaid node definition.

    Uses the v11.3.0+ ``@{ shape: <name> }`` syntax when a shape is set,
    falling back to the bracket syntax (``["name"]``) otherwise.

    Args:
        use_alias: If True, use the alias instead of full ID
        indentation: Number of spaces to indent
        show_instance: If True, include domo_instance on line 3
        show_entity_id: If True, include entity ID on line 4
    """
    label = self._build_label(
        show_instance=show_instance,
        show_entity_id=show_entity_id,
    )
    safe_label = label.replace('"', "'").replace("\n", " ")

    if use_alias and self.alias:
        node_id = self.alias
    else:
        node_id = f"{self.type}_{self.id}".replace("-", "_").replace(" ", "_")[:50]

    # Resolve shape: explicit > entity-type mapping > default (rect)
    shape = self.shape or self.ENTITY_TYPE_SHAPES.get(self.type)

    if shape:
        return f'{" " * indentation}{node_id}@{{ shape: {shape}, label: "{safe_label}" }}'

    return f'{" " * indentation}{node_id}["{safe_label}"]'

MermaidRelationship dataclass

MermaidRelationship(
    from_node: MermaidNode,
    to_node: MermaidNode,
    type: str = "DEPENDS_ON",
)

Represents a relationship between two nodes in a Mermaid diagram.

to_string

to_string(
    use_alias: bool = False, indentation: int = 4
) -> str

Render the relationship as a Mermaid arrow with type label.

All relationships render as A -->|TYPE| B including DEPENDS_ON.

Source code in src/crew_dcs/integrations/graphs/mermaid/mermaid_relationship.py

def to_string(self, use_alias: bool = False, indentation: int = 4) -> str:
    """Render the relationship as a Mermaid arrow with type label.

    All relationships render as ``A -->|TYPE| B`` including DEPENDS_ON.
    """
    if use_alias and self.from_node.alias:
        from_id = self.from_node.alias
    else:
        from_id = f"{self.from_node.type}_{self.from_node.id}".replace(
            "-", "_"
        ).replace(" ", "_")[:50]

    if use_alias and self.to_node.alias:
        to_id = self.to_node.alias
    else:
        to_id = f"{self.to_node.type}_{self.to_node.id}".replace("-", "_").replace(
            " ", "_"
        )[:50]

    rel_type = self.type or "DEPENDS_ON"
    return f"{' ' * indentation}{from_id} -->|{rel_type}| {to_id}"

Neo4jConverter

Bases: GraphConverter

Convert lineage graph to Neo4j Cypher queries.

convert

convert(graph: LineageGraph, **options) -> list[str]

Convert LineageGraph to Cypher CREATE queries.

Parameters:

Name	Type	Description	Default
graph	LineageGraph	LineageGraph with nodes and edges	required
**options		Additional options (unused)	{}

Returns:

Type	Description
list[str]	List of Cypher queries (nodes first, then relationships)

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

def convert(
    self,
    graph: LineageGraph,
    **options,
) -> list[str]:
    """Convert LineageGraph to Cypher CREATE queries.

    Args:
        graph: LineageGraph with nodes and edges
        **options: Additional options (unused)

    Returns:
        List of Cypher queries (nodes first, then relationships)
    """
    queries: list[str] = []

    # Generate node creation queries
    queries.append("// Create nodes")
    for node in graph.nodes:
        labels = f":{node.entity_type}:DomoEntity"
        props = self._format_properties(node)
        query = f"CREATE (n{labels} {props})"
        queries.append(query)

    queries.append("")
    queries.append("// Create relationships")

    # Generate relationship creation queries
    for edge in graph.edges:
        query = self._build_relationship_query(edge)
        queries.append(query)

    return queries

export_to_file

export_to_file(
    output: list[str], file_path: str, **options
) -> None

Export Cypher queries to .cypher file.

Parameters:

Name	Type	Description	Default
output	list[str]	List of Cypher queries from convert()	required
file_path	str	Destination file path	required
**options		Additional options (unused)	{}

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

def export_to_file(
    self,
    output: list[str],
    file_path: str,
    **options,
) -> None:
    """Export Cypher queries to .cypher file.

    Args:
        output: List of Cypher queries from convert()
        file_path: Destination file path
        **options: Additional options (unused)
    """
    with open(file_path, "w", encoding="utf-8") as f:
        for query in output:
            f.write(query)
            if not query.startswith("//"):
                f.write(";")
            f.write("\n\n")

Modules

• converter
• graph
• mermaid
• neo4j_converter