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.execute_query("MATCH (n) DETACH DELETE n")

execute_query

execute_query(query: str, params: dict | None = None) -> 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

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,
) -> list[dict]:
    """Execute a Cypher query against the database.

    Args:
        query: The Cypher query string.
        params: Optional query parameters.

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

execute_query_as_objects

execute_query_as_objects(query: str, params: dict | None = None, node_id_to_object: 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

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.

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,
) -> 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.

    Returns:
        A list of rows, where each row is a list of Python objects
        converted from Node instances in the query results.
    """
    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
    )
    for row_dict in results:
        row = [
            fieldz_kb.lpg.core.make_object_from_node(
                self._context, value, node_id_to_object
            )
            for value in row_dict.values()
            if isinstance(value, pylpg.node.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) -> 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

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,
) -> 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.
    """
    self.save_from_objects(
        objects=[object_],
        integration_mode=integration_mode,
        exclude_from_integration=exclude_from_integration,
    )

save_from_objects

save_from_objects(objects: list[object], integration_mode: Literal['hash', 'id'] = 'id', exclude_from_integration: tuple[type, ...] | 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

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,
) -> 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.

    Raises:
        ValueError: If a node is not a subclass of BaseNode.
    """
    if exclude_from_integration is None:
        exclude_from_integration = tuple()
    object_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_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)