Session

pylpg.session

Session for orchestrating object persistence.

Classes:

Name	Description
Session	Orchestrates saving, deleting, and querying graph objects.

Session

Session(backend: Backend)

Orchestrates saving, deleting, and querying graph objects.

Use as a context manager to ensure the backend connection is closed.

Example

with Session(backend) as session:
    alice = Person(name="Alice")
    session.save(alice)

Methods:

Name	Description
delete	Delete a node, relationship, or a list of both.
delete_all	Delete every node and relationship in the database.
execute_query	Execute a raw Cypher query.
prefetch	Batch-load the subgraph reachable from roots into memory.
save	Save a node, relationship, or a list of both.

Source code in src/pylpg/session.py

def __init__(self, backend: pylpg.backend.base.Backend) -> None:
    self._backend = backend
    self._prefetch: dict[typing.Any, _NodeEdges] | None = None

delete

delete(item: Node | Relationship | list[Node | Relationship]) -> None

Delete a node, relationship, or a list of both.

Example

session.delete(alice)
session.delete([alice, bob])

Source code in src/pylpg/session.py

def delete(
    self,
    item: (
        pylpg.node.Node
        | pylpg.relationship.Relationship
        | list[pylpg.node.Node | pylpg.relationship.Relationship]
    ),
) -> None:
    """Delete a node, relationship, or a list of both.

    Example:
        ```python
        session.delete(alice)
        session.delete([alice, bob])
        ```
    """
    self._prefetch = None
    if isinstance(item, pylpg.node.Node):
        self._delete_node(item)
    elif isinstance(item, pylpg.relationship.Relationship):
        self._delete_relationship(item)
    elif isinstance(item, list):
        for element in item:
            self.delete(element)

delete_all

delete_all() -> None

Delete every node and relationship in the database.

Delegates to the backend so each backend can apply its own strategy (e.g. FalkorDB drops the graph to clear stale label indexes that survive a plain MATCH (n) DETACH DELETE n).

Example

session.delete_all()

Source code in src/pylpg/session.py

def delete_all(self) -> None:
    """Delete every node and relationship in the database.

    Delegates to the backend so each backend can apply its own
    strategy (e.g. FalkorDB drops the graph to clear stale label
    indexes that survive a plain `MATCH (n) DETACH DELETE n`).

    Example:
        ```python
        session.delete_all()
        ```
    """
    self._prefetch = None
    self._backend.delete_all()

execute_query

execute_query(cypher: str, parameters: dict[str, Any] | None = None, resolve_nodes: bool = False) -> list[dict[str, Any]]

Execute a raw Cypher query.

When resolve_nodes is True, any driver node objects in the results are hydrated into Node instances.

Example

results = session.execute_query(
    "MATCH (n:Person) RETURN n",
    resolve_nodes=True,
)
for row in results:
    person = row["n"]  # a Person instance

Source code in src/pylpg/session.py

def execute_query(
    self,
    cypher: str,
    parameters: dict[str, typing.Any] | None = None,
    resolve_nodes: bool = False,
) -> list[dict[str, typing.Any]]:
    """Execute a raw Cypher query.

    When `resolve_nodes` is True, any driver node objects in the
    results are hydrated into `Node` instances.

    Example:
        ```python
        results = session.execute_query(
            "MATCH (n:Person) RETURN n",
            resolve_nodes=True,
        )
        for row in results:
            person = row["n"]  # a Person instance
        ```
    """
    results = self._backend.execute_query(cypher, parameters)
    if resolve_nodes:
        return [self._hydrate_row(row) for row in results]
    return results

prefetch

prefetch(roots: list[Node]) -> Iterator[None]

Batch-load the subgraph reachable from roots into memory.

Inside the block, traversal reads the snapshot instead of querying: one query per (relationship type, direction) per level, rather than one per node per field. Saving or deleting inside the block drops the snapshot; writes made through raw execute_query do not.

Example

with session.prefetch(roots=[alice, bob]):
    for friend in alice.friends.all():  # no query
        ...

Source code in src/pylpg/session.py

@contextlib.contextmanager
def prefetch(self, roots: list[pylpg.node.Node]) -> typing.Iterator[None]:
    """Batch-load the subgraph reachable from `roots` into memory.

    Inside the block, traversal reads the snapshot instead of querying:
    one query per (relationship type, direction) per level, rather than
    one per node per field. Saving or deleting inside the block drops the
    snapshot; writes made through raw `execute_query` do not.

    Example:
        ```python
        with session.prefetch(roots=[alice, bob]):
            for friend in alice.friends.all():  # no query
                ...
        ```
    """
    frontier = {node._database_id: node for node in roots if node.is_saved()}
    cache: dict[typing.Any, _NodeEdges] = {
        database_id: {} for database_id in frontier
    }
    while frontier:
        sources_by_key: dict[
            tuple[str, pylpg.relationship.Direction], set[typing.Any]
        ] = {}
        for node in frontier.values():
            for descriptor in type(node).__relationship_descriptors__.values():
                key = (
                    descriptor._relationship_class.__type__,
                    descriptor._direction,
                )
                sources_by_key.setdefault(key, set()).add(node._database_id)
        next_frontier: dict[typing.Any, pylpg.node.Node] = {}
        for key, source_ids in sources_by_key.items():
            (relationship_type, direction) = key
            rows = self._backend.traverse_batch(
                source_ids=list(source_ids),
                relationship_type=relationship_type,
                direction=direction,
            )
            for row in rows:
                target = self._hydrate_node(
                    deserialized_node=self._backend.deserialize_node(
                        record=row["target"]
                    )
                )
                relationship = self._backend.deserialize_relationship(
                    record=row["relationship"]
                )
                cache[row["source_id"]].setdefault(key, []).append(
                    (target, relationship)
                )
                if target._database_id not in cache:
                    cache[target._database_id] = {}
                    next_frontier[target._database_id] = target
        frontier = next_frontier
    self._prefetch = cache
    try:
        yield
    finally:
        self._prefetch = None

save

save(item: Node | Relationship | list[Node | Relationship]) -> None

Save a node, relationship, or a list of both.

For single items, creates or updates depending on whether the item has already been saved. For lists, delegates to the backend's batch save strategy.

Example

session.save(alice)
session.save(relationship)
session.save([alice, bob, relationship])

Source code in src/pylpg/session.py

def save(
    self,
    item: (
        pylpg.node.Node
        | pylpg.relationship.Relationship
        | list[pylpg.node.Node | pylpg.relationship.Relationship]
    ),
) -> None:
    """Save a node, relationship, or a list of both.

    For single items, creates or updates depending on whether the
    item has already been saved. For lists, delegates to the
    backend's batch save strategy.

    Example:
        ```python
        session.save(alice)
        session.save(relationship)
        session.save([alice, bob, relationship])
        ```
    """
    self._prefetch = None
    if isinstance(item, pylpg.node.Node):
        self._save_node(item)
    elif isinstance(item, pylpg.relationship.Relationship):
        self._save_relationship(item)
    elif isinstance(item, list):
        self._backend.save_batch(items=item)
        for element in item:
            if isinstance(element, pylpg.node.Node):
                element._session = self
            elif isinstance(element, pylpg.relationship.Relationship):
                element.source._session = self
                element.target._session = self