Skip to content

neo4j_converter

neo4j_converter

Neo4j Cypher query converter for lineage graphs.

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 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
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 
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
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")