Skip to content

FalkorDB backend

pylpg.backend.falkordb

FalkorDB backend using the FalkorDB Python driver.

Classes:

Name Description
FalkorDBBackend

Backend for FalkorDB server instances.

FalkorDBBackend 

FalkorDBBackend(hostname: str = 'localhost', port: int = 6379, database: str = 'default', username: str | None = None, password: str | None = None)

Bases: Backend

Backend for FalkorDB server instances.

Uses individual queries for batch operations, which is optimal for FalkorDB's in-memory architecture.

Methods:

Name Description
deserialize_node

Return a node record as a dict of its properties.
deserialize_relationship

Return a relationship record as a dict of its properties.
result_set_limit

Maximum number of rows a single query may return.
traverse_batch

Traverse from many source nodes at once.
Source code in src/pylpg/backend/falkordb.py 
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    hostname: str = "localhost",
    port: int = 6379,
    database: str = "default",
    username: str | None = None,
    password: str | None = None,
) -> None:
    self._db = falkordb.FalkorDB(
        host=hostname, port=port, username=username, password=password
    )
    self._graph = self._db.select_graph(database)

deserialize_node 

deserialize_node(record: Any) -> dict[str, Any]

Return a node record as a dict of its properties.

The dict also carries _labels and _database_id.

Source code in src/pylpg/backend/falkordb.py 
62
63
64
65
66
def deserialize_node(self, record: typing.Any) -> dict[str, typing.Any]:
    properties = dict(record.properties)
    properties["_labels"] = frozenset(record.labels)
    properties["_database_id"] = record.id
    return properties

deserialize_relationship 

deserialize_relationship(record: Any) -> dict[str, Any]

Return a relationship record as a dict of its properties.

The dict also carries _database_id, _start_id and _end_id. The record must come from a matched pattern, not from a path.

Source code in src/pylpg/backend/falkordb.py 
68
69
70
71
72
73
def deserialize_relationship(self, record: typing.Any) -> dict[str, typing.Any]:
    properties = dict(record.properties)
    properties["_database_id"] = record.id
    properties["_start_id"] = record.src_node
    properties["_end_id"] = record.dest_node
    return properties

result_set_limit 

result_set_limit() -> int | None

Maximum number of rows a single query may return.

Returns None when the backend imposes no limit. Backends that silently truncate oversized result sets (FalkorDB caps at RESULTSET_SIZE, 10000 by default) must report their limit here so traverse_batch can split batches instead of losing rows.

Source code in src/pylpg/backend/falkordb.py 
37
38
39
40
41
42
43
44
45
46
47
48
def result_set_limit(self) -> int | None:
    # FalkorDB truncates any result set to RESULTSET_SIZE rows without
    # raising, so batched traversal must know the limit. A non-positive
    # value means unlimited. Not cached: the read costs ~47us, well under
    # a trivial round trip, and reading it every time picks up a runtime
    # GRAPH.CONFIG SET without needing a new session.
    # A server too old to know RESULTSET_SIZE has no cap to report.
    try:
        value = int(self._db.config_get("RESULTSET_SIZE"))
    except (redis.exceptions.ResponseError, TypeError, ValueError):
        value = -1
    return value if value > 0 else None

traverse_batch 

traverse_batch(source_ids: list[Any], relationship_type: str, direction: Direction) -> list[dict[str, Any]]

Traverse from many source nodes at once.

Splits the batch and retries whenever a result set comes back at the backend's row limit, since such a result set may have been silently truncated.

Source code in src/pylpg/backend/base.py 
 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
122
123
124
125
126
127
128
129
130
131
132
133
def traverse_batch(
    self,
    source_ids: list[typing.Any],
    relationship_type: str,
    direction: "pylpg.relationship.Direction",
) -> list[dict[str, typing.Any]]:
    """Traverse from many source nodes at once.

    Splits the batch and retries whenever a result set comes back at
    the backend's row limit, since such a result set may have been
    silently truncated.
    """
    limit = self.result_set_limit()
    rows: list[dict[str, typing.Any]] = []
    pending = [list(source_ids)]
    while pending:
        chunk = pending.pop()
        if not chunk:
            continue
        chunk_rows = self._traverse_batch_chunk(
            source_ids=chunk,
            relationship_type=relationship_type,
            direction=direction,
        )
        if limit is not None and len(chunk_rows) >= limit:
            if len(chunk) == 1:
                raise ValueError(
                    f"Node {chunk[0]} has at least {limit} '{relationship_type}' "
                    f"relationships, which reaches this backend's result set "
                    f"limit of {limit} rows. The result would be silently "
                    f"truncated. Raise the backend's limit (for FalkorDB: "
                    f"GRAPH.CONFIG SET RESULTSET_SIZE) to traverse this node."
                )
            middle = len(chunk) // 2
            pending.append(chunk[:middle])
            pending.append(chunk[middle:])
            continue
        rows.extend(chunk_rows)
    return rows