Skip to content

graflo.db.cypher

Shared Cypher query fragments (no drivers; safe string builders only).

cypher_map_key(name)

Return a backtick-quoted map-key / property name for Cypher patterns.

Strips embedded backticks from name so callers cannot break out of quotes.

Source code in graflo/db/cypher/escape.py 
11
12
13
14
15
16
17
18
19
def cypher_map_key(name: str) -> str:
    """Return a backtick-quoted map-key / property name for Cypher patterns.

    Strips embedded backticks from *name* so callers cannot break out of quotes.
    """
    key = name.strip().replace("`", "")
    if not key:
        raise ValueError("Cypher property name must be non-empty")
    return f"`{key}`"

cypher_string_literal(value)

Return a single-quoted Cypher string literal with escapes.

Source code in graflo/db/cypher/escape.py 
6
7
8
def cypher_string_literal(value: str) -> str:
    """Return a single-quoted Cypher string literal with escapes."""
    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"

rel_merge_props_map_from_row_index(prop_names, *, row_index=2)

Build `k`: row[n]['k'], ... for MERGE relationship properties.

Matches batches shaped as row = [source_doc, target_doc, props] (Neo4j, FalkorDB-style row[2]).

Source code in graflo/db/cypher/rel_merge.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def rel_merge_props_map_from_row_index(
    prop_names: Sequence[str], *, row_index: int = 2
) -> str:
    """Build `` `k`: row[n]['k'], ... `` for MERGE relationship properties.

    Matches batches shaped as ``row`` = ``[source_doc, target_doc, props]`` (Neo4j,
    FalkorDB-style ``row[2]``).
    """
    row_access = f"row[{row_index}]"
    parts: list[str] = []
    for key in _normalized_prop_names(prop_names):
        bk = cypher_map_key(key)
        lit = cypher_string_literal(key)
        parts.append(f"{bk}: {row_access}[{lit}]")
    return ", ".join(parts)

rel_merge_props_map_from_row_props(prop_names, *, props_expr='row.props')

Build `k`: row.props['k'], ... (Memgraph-style batch rows).

Source code in graflo/db/cypher/rel_merge.py 
36
37
38
39
40
41
42
43
44
45
def rel_merge_props_map_from_row_props(
    prop_names: Sequence[str], *, props_expr: str = "row.props"
) -> str:
    """Build `` `k`: row.props['k'], ... `` (Memgraph-style batch rows)."""
    parts: list[str] = []
    for key in _normalized_prop_names(prop_names):
        bk = cypher_map_key(key)
        lit = cypher_string_literal(key)
        parts.append(f"{bk}: {props_expr}[{lit}]")
    return ", ".join(parts)