mermaid_diagram

mermaid_diagram

MermaidDiagram — complete Mermaid diagram with nodes, relationships, and export.

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)