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.
execute_query	Execute a raw Cypher query.
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

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])
        ```
    """
    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)

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

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])
        ```
    """
    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