Skip to content

mermaid_converter

mermaid_converter

MermaidConverter — converts LineageGraph to MermaidDiagram via GraphConverter ABC.

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
20
21
22
23
24
25
26
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
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
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
123
124
125
126
127
128
129
130
131
132
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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
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
44
45
46
47
48
49
50
51
52
53
54
@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(
    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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
@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)