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

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_neighbors_query(*, anchor_label, anchor_id, anchor_key_field='id', edge_type, far_label, direction, hops, limit)

Render a bounded neighbourhood query.

Returns the reached nodes and their distance, deduplicated. DISTINCT is load-bearing: a graph with a cycle reaches the same node by several paths, and without it the row count grows with path multiplicity rather than with neighbourhood size.

Source code in graflo/db/cypher/traversal.py

def cypher_neighbors_query(
    *,
    anchor_label: str,
    anchor_id: str,
    anchor_key_field: str = "id",
    edge_type: str | None,
    far_label: str | None,
    direction: EdgeDirection,
    hops: int,
    limit: int | None,
) -> str:
    """Render a bounded neighbourhood query.

    Returns the reached nodes and their distance, deduplicated. ``DISTINCT`` is
    load-bearing: a graph with a cycle reaches the same node by several paths,
    and without it the row count grows with path multiplicity rather than with
    neighbourhood size.
    """
    if hops < 1:
        raise ValueError(f"hops must be >= 1, got {hops}")
    pattern = cypher_rel_pattern(edge_type, direction, min_hops=1, max_hops=hops)
    far = f"(far:{far_label})" if far_label else "(far)"
    limit_clause = f"\nLIMIT {int(limit)}" if limit is not None else ""
    return (
        f"MATCH path = (anchor:{anchor_label} "
        f"{{{anchor_key_field}: '{anchor_id}'}}){pattern}{far}\n"
        f"RETURN DISTINCT properties(far) AS far, length(path) AS distance"
        f"{limit_clause}"
    )

cypher_rel_pattern(edge_type, direction=EdgeDirection.OUT, *, variable='r', min_hops=None, max_hops=None)

Render the relationship pattern between two node patterns.

Parameters:

Name	Type	Description	Default
edge_type	str | None	Relationship type to filter on, or None for any type.	required
direction	EdgeDirection	Orientation followed from the left-hand (anchor) node.	OUT
variable	str	Relationship variable name bound in the pattern.	'r'
min_hops	int | None	Lower bound for a variable-length pattern. Defaults to 1 when only max_hops is given.	None
max_hops	int | None	Upper bound for a variable-length pattern. Omitting both keeps the single-hop form, which is what every existing caller wants.	None

Returns:

Name	Type	Description
str	str	e.g. -[r:KNOWS]-> (OUT), <-[r:KNOWS]- (IN),
	str	-[r:KNOWS]- (ANY), -[r:KNOWS*1..3]-> (variable length).

Raises:

Type	Description
ValueError	if the hop bounds are non-positive or inverted — an unbounded * pattern is a full-graph scan and is never emitted.

Source code in graflo/db/cypher/direction.py

def cypher_rel_pattern(
    edge_type: str | None,
    direction: EdgeDirection = EdgeDirection.OUT,
    *,
    variable: str = "r",
    min_hops: int | None = None,
    max_hops: int | None = None,
) -> str:
    """Render the relationship pattern between two node patterns.

    Args:
        edge_type: Relationship type to filter on, or None for any type.
        direction: Orientation followed from the left-hand (anchor) node.
        variable: Relationship variable name bound in the pattern.
        min_hops: Lower bound for a variable-length pattern. Defaults to 1 when
            only *max_hops* is given.
        max_hops: Upper bound for a variable-length pattern. Omitting both keeps
            the single-hop form, which is what every existing caller wants.

    Returns:
        str: e.g. ``-[r:KNOWS]->`` (OUT), ``<-[r:KNOWS]-`` (IN),
        ``-[r:KNOWS]-`` (ANY), ``-[r:KNOWS*1..3]->`` (variable length).

    Raises:
        ValueError: if the hop bounds are non-positive or inverted — an
            unbounded ``*`` pattern is a full-graph scan and is never emitted.
    """
    left, right = _ARROWS[direction]
    quantifier = _hop_quantifier(min_hops, max_hops)
    body = (
        f"[{variable}:{edge_type}{quantifier}]"
        if edge_type
        else f"[{variable}{quantifier}]"
    )
    return f"{left}{body}{right}"

cypher_string_literal(value)

Return a single-quoted Cypher string literal with escapes.

Source code in graflo/db/cypher/escape.py

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

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

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)