Session

fieldz_kb.lpg.session

Pylpg session for fieldz_kb.

Provides a Session class that wraps a pylpg.Session and adds higher-level methods for saving and retrieving fieldz objects.

Classes:

Name	Description
Session	Session for saving and retrieving fieldz objects via pylpg.

Session

Session(backend: Backend)

Session for saving and retrieving fieldz objects via pylpg.

Wraps a pylpg.Session internally. Use as a context manager.

Example

import fieldz_kb.lpg import fieldz_kb.lpg.backends.neo4j backend = fieldz_kb.lpg.backends.neo4j.Neo4jBackend(hostname="localhost") with fieldz_kb.lpg.session.Session(backend) as session: ... session.save_from_object(person)

Initialize the session with a pylpg backend.

Parameters:

Name	Type	Description	Default
backend	Backend	A pylpg backend instance (Neo4jBackend, FalkorDBBackend, etc.)	required

Methods:

Name	Description
__enter__	Enter the session context manager.
__exit__	Exit the session context manager.
delete_all	Delete all nodes and relationships from the database.
execute_query	Execute a Cypher query against the database.
execute_query_as_objects	Execute a Cypher query and convert results to Python objects.
reset_context	Replace the conversion context with a fresh one.
save_from_object	Save a single object to the database.
save_from_objects	Save multiple objects to the database.

Source code in src/fieldz_kb/lpg/session.py

def __init__(self, backend: pylpg.backend.base.Backend) -> None:
    """Initialize the session with a pylpg backend.

    Args:
        backend: A pylpg backend instance (Neo4jBackend, FalkorDBBackend, etc.)
    """
    self._pylpg_session = pylpg.session.Session(backend)
    self._context = fieldz_kb.lpg.core.get_default_context()

__enter__

__enter__() -> Session

Enter the session context manager.

Source code in src/fieldz_kb/lpg/session.py

def __enter__(self) -> "Session":
    """Enter the session context manager."""
    self._pylpg_session.__enter__()
    return self

__exit__

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: object) -> bool | None

Exit the session context manager.

Source code in src/fieldz_kb/lpg/session.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: object,
) -> bool | None:
    """Exit the session context manager."""
    return self._pylpg_session.__exit__(exc_type, exc_val, exc_tb)

delete_all

delete_all() -> None

Delete all nodes and relationships from the database.

Source code in src/fieldz_kb/lpg/session.py

def delete_all(self) -> None:
    """Delete all nodes and relationships from the database."""
    self._pylpg_session.delete_all()

execute_query

execute_query(query: str, params: dict | None = None, resolve_nodes: bool = False) -> list[dict]

Execute a Cypher query against the database.

Parameters:

Name	Type	Description	Default
query	str	The Cypher query string.	required
params	dict | None	Optional query parameters.	None
resolve_nodes	bool	When True, any driver node objects in the results are hydrated into pylpg.node.Node instances.	False

Returns:

Type	Description
list[dict]	Query results as a list of dicts.

Source code in src/fieldz_kb/lpg/session.py

def execute_query(
    self,
    query: str,
    params: dict | None = None,
    resolve_nodes: bool = False,
) -> list[dict]:
    """Execute a Cypher query against the database.

    Args:
        query: The Cypher query string.
        params: Optional query parameters.
        resolve_nodes: When True, any driver node objects in the results
            are hydrated into `pylpg.node.Node` instances.

    Returns:
        Query results as a list of dicts.
    """
    return self._pylpg_session.execute_query(
        query, parameters=params, resolve_nodes=resolve_nodes
    )

execute_query_as_objects

execute_query_as_objects(query: str, params: dict | None = None, node_id_to_object: dict | None = None, object_key_to_node: dict | None = None) -> list[list[object]]

Execute a Cypher query and convert results to Python objects.

Parameters:

Name	Type	Description	Default
query	str	The Cypher query string.	required
params	dict | None	Optional query parameters.	None
node_id_to_object	dict | None	Optional cache mapping node database IDs to objects.	None
object_key_to_node	dict | None	Optional cache mapping objects to the nodes they were converted from, populated in place. Pass it on to a subsequent save to update the existing nodes instead of creating duplicates. Only the objects a row returns directly are recorded, not the ones nested inside them, so a query must return whatever it needs seeded. Keys are the objects themselves, which matches "hash" integration mode only; a save under "id" mode keys on object ids and misses every entry.	None

Returns:

Type	Description
list[list[object]]	A list of rows, where each row is a list of Python objects
list[list[object]]	converted from Node instances in the query results.

Raises:

Type	Description
ValueError	If object_key_to_node is given and a converted object is not hashable.

Source code in src/fieldz_kb/lpg/session.py

def execute_query_as_objects(
    self,
    query: str,
    params: dict | None = None,
    node_id_to_object: dict | None = None,
    object_key_to_node: dict | None = None,
) -> list[list[object]]:
    """Execute a Cypher query and convert results to Python objects.

    Args:
        query: The Cypher query string.
        params: Optional query parameters.
        node_id_to_object: Optional cache mapping node database IDs to objects.
        object_key_to_node: Optional cache mapping objects to the nodes they
            were converted from, populated in place. Pass it on to a
            subsequent save to update the existing nodes instead of creating
            duplicates. Only the objects a row returns directly are recorded,
            not the ones nested inside them, so a query must return whatever
            it needs seeded. Keys are the objects themselves, which matches
            "hash" integration mode only; a save under "id" mode keys on
            object ids and misses every entry.

    Returns:
        A list of rows, where each row is a list of Python objects
        converted from Node instances in the query results.

    Raises:
        ValueError: If object_key_to_node is given and a converted object is
            not hashable.
    """
    if node_id_to_object is None:
        node_id_to_object = {}
    object_results = []
    results = self._pylpg_session.execute_query(
        query, parameters=params, resolve_nodes=True
    )
    root_nodes = [
        value
        for row_dict in results
        for value in row_dict.values()
        if isinstance(value, pylpg.node.Node)
    ]
    with self._pylpg_session.prefetch(roots=root_nodes):
        for row_dict in results:
            row_nodes = [
                value
                for value in row_dict.values()
                if isinstance(value, pylpg.node.Node)
            ]
            row = [
                fieldz_kb.lpg.core.make_object_from_node(
                    self._context, node, node_id_to_object
                )
                for node in row_nodes
            ]
            if object_key_to_node is not None:
                for node, object_ in zip(row_nodes, row):
                    if not isinstance(object_, collections.abc.Hashable):
                        raise ValueError(
                            f"object of type {type(object_)} not hashable, "
                            "cannot populate object_key_to_node"
                        )
                    object_key_to_node[object_] = node
            object_results.append(row)
    return object_results

reset_context

reset_context() -> None

Replace the conversion context with a fresh one.

Clears all cached node classes and type mappings while keeping the same database connection.

Source code in src/fieldz_kb/lpg/session.py

def reset_context(self) -> None:
    """Replace the conversion context with a fresh one.

    Clears all cached node classes and type mappings while keeping the
    same database connection.
    """
    self._context = fieldz_kb.lpg.core.make_context()

save_from_object

save_from_object(object_: object, integration_mode: Literal['hash', 'id'] = 'id', exclude_from_integration: tuple[type, ...] | None = None, object_key_to_node: dict | None = None) -> None

Save a single object to the database.

Parameters:

Name	Type	Description	Default
object_	object	The object to save.	required
integration_mode	Literal['hash', 'id']	How to handle duplicate objects ("hash" or "id").	'id'
exclude_from_integration	tuple[type, ...] | None	Types to exclude from integration logic.	None
object_key_to_node	dict | None	Optional cache mapping object keys to their nodes, updated in place. Pass the same dict across calls to integrate objects shared between them. Keys are the objects themselves under "hash" integration mode and their ids under "id" mode, so a shared cache is only valid while the mode is constant.	None

Source code in src/fieldz_kb/lpg/session.py

def save_from_object(
    self,
    object_: object,
    integration_mode: typing.Literal["hash", "id"] = "id",
    exclude_from_integration: tuple[type, ...] | None = None,
    object_key_to_node: dict | None = None,
) -> None:
    """Save a single object to the database.

    Args:
        object_: The object to save.
        integration_mode: How to handle duplicate objects ("hash" or "id").
        exclude_from_integration: Types to exclude from integration logic.
        object_key_to_node: Optional cache mapping object keys to their
            nodes, updated in place. Pass the same dict across calls to
            integrate objects shared between them. Keys are the objects
            themselves under "hash" integration mode and their ids under
            "id" mode, so a shared cache is only valid while the mode is
            constant.
    """
    self.save_from_objects(
        objects=[object_],
        integration_mode=integration_mode,
        exclude_from_integration=exclude_from_integration,
        object_key_to_node=object_key_to_node,
    )

save_from_objects

save_from_objects(objects: list[object], integration_mode: Literal['hash', 'id'] = 'id', exclude_from_integration: tuple[type, ...] | None = None, object_key_to_node: dict | None = None) -> None

Save multiple objects to the database.

Parameters:

Name	Type	Description	Default
objects	list[object]	The objects to save.	required
integration_mode	Literal['hash', 'id']	How to handle duplicate objects ("hash" or "id").	'id'
exclude_from_integration	tuple[type, ...] | None	Types to exclude from integration logic.	None
object_key_to_node	dict | None	Optional cache mapping object keys to their nodes, updated in place. Pass the same dict across calls to integrate objects shared between them. Keys are the objects themselves under "hash" integration mode and their ids under "id" mode, so a shared cache is only valid while the mode is constant.	None

Raises:

Type	Description
ValueError	If a node is not a subclass of BaseNode.

Source code in src/fieldz_kb/lpg/session.py

def save_from_objects(
    self,
    objects: list[object],
    integration_mode: typing.Literal["hash", "id"] = "id",
    exclude_from_integration: tuple[type, ...] | None = None,
    object_key_to_node: dict | None = None,
) -> None:
    """Save multiple objects to the database.

    Args:
        objects: The objects to save.
        integration_mode: How to handle duplicate objects ("hash" or "id").
        exclude_from_integration: Types to exclude from integration logic.
        object_key_to_node: Optional cache mapping object keys to their
            nodes, updated in place. Pass the same dict across calls to
            integrate objects shared between them. Keys are the objects
            themselves under "hash" integration mode and their ids under
            "id" mode, so a shared cache is only valid while the mode is
            constant.

    Raises:
        ValueError: If a node is not a subclass of BaseNode.
    """
    if exclude_from_integration is None:
        exclude_from_integration = tuple()
    if object_key_to_node is None:
        object_key_to_node = {}
    saved_node_ids = set()
    all_nodes = []
    all_relationships = []
    for object_ in objects:
        nodes, relationships = fieldz_kb.lpg.core.make_nodes_from_object(
            self._context,
            object_,
            integration_mode,
            exclude_from_integration,
            object_key_to_node,
        )
        for node in nodes:
            if id(node) not in saved_node_ids:
                if not isinstance(node, fieldz_kb.lpg.graph.BaseNode):
                    raise ValueError(
                        f"node type {type(node)} must be a subclass of BaseNode"
                    )
                all_nodes.append(node)
                saved_node_ids.add(id(node))
        all_relationships += relationships
    self._pylpg_session.save(all_nodes)
    self._pylpg_session.save(all_relationships)