Neo4j backend

pylpg.backend.neo4j

Neo4j backend using the official Neo4j Python driver.

Classes:

Name	Description
Neo4jBackend	Backend for Neo4j databases.

Neo4jBackend

Neo4jBackend(hostname: str = 'localhost', port: int = 7687, database: str = 'neo4j', username: str = 'neo4j', password: str = 'neo4j', protocol: str = 'bolt', notifications_min_severity: Literal['off', 'information', 'warning'] | None = None)

Bases: Backend

Backend for Neo4j databases.

Uses UNWIND queries for batch operations, providing significant speedups over individual queries at scale.

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/neo4j.py

def __init__(
    self,
    hostname: str = "localhost",
    port: int = 7687,
    database: str = "neo4j",
    username: str = "neo4j",
    password: str = "neo4j",
    protocol: str = "bolt",
    notifications_min_severity: typing.Literal["off", "information", "warning"]
    | None = None,
) -> None:
    uri = f"{protocol}://{hostname}:{port}"
    driver_kwargs: dict[str, typing.Any] = {"auth": (username, password)}
    if notifications_min_severity is not None:
        driver_kwargs["notifications_min_severity"] = (
            notifications_min_severity.upper()
        )
    self._driver = neo4j.GraphDatabase.driver(uri, **driver_kwargs)
    self._database = 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/neo4j.py

def deserialize_node(self, record: typing.Any) -> dict[str, typing.Any]:
    properties = dict(record)
    properties["_labels"] = frozenset(record.labels)
    properties["_database_id"] = record.element_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/neo4j.py

def deserialize_relationship(self, record: typing.Any) -> dict[str, typing.Any]:
    properties = dict(record)
    properties["_database_id"] = record.element_id
    properties["_start_id"] = record.start_node.element_id
    properties["_end_id"] = record.end_node.element_id
    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/base.py

def result_set_limit(self) -> 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.
    """
    return 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

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