Session

momapy_kb.lpg.session

LPG session for momapy_kb.

Provides a Session class that wraps a fieldz_kb LPG session and adds momapy-specific functionality: file loading, collection management, and layout element queries.

Classes:

Name	Description
Session	Session for saving and retrieving momapy objects via a graph database.

Session

Session(backend)

Session for saving and retrieving momapy objects via a graph database.

Wraps a fieldz_kb LPG session and registers momapy-specific type plugins. Use as a context manager.

Parameters:

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

Example

import momapy_kb.lpg.session import momapy_kb.lpg.backends.neo4j backend = momapy_kb.lpg.backends.neo4j.Neo4jBackend(hostname="localhost") with momapy_kb.lpg.session.Session(backend) as session: ... session.save_from_file("model.xml", integration_mode="hash")

Methods:

Name	Description
delete_all	Delete all nodes and relationships from the database.
execute_query	Execute a Cypher query against the database.
execute_query_as_layout_elements	Execute a Cypher query and return results as layout elements.
execute_query_as_objects	Execute a Cypher query and convert results to Python objects.
get_layout_element_nodes_from_model_element_node	Get layout element nodes linked to a model element node.
make_layout_elements_from_model_element_node	Get layout elements as Python objects for a model element node.
save_collections_from_entries	Save collections from pre-built CollectionEntry objects.
save_collections_from_file_paths	Load maps from files and save them as named collections.
save_from_file	Load a map from a file and save it to the database.
save_from_object	Save a single object to the database.
save_from_objects	Save multiple objects to the database.

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

def __init__(self, backend) -> None:
    self._session = fieldz_kb.lpg.session.Session(backend)
    momapy_kb.lpg.types.register_momapy_plugins(self._session._context)

delete_all

delete_all() -> None

Delete all nodes and relationships from the database.

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

def delete_all(self) -> None:
    """Delete all nodes and relationships from the database."""
    self._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/momapy_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._session.execute_query(
        query, params=params, resolve_nodes=resolve_nodes
    )

execute_query_as_layout_elements

execute_query_as_layout_elements(query: str, params: dict | None = None) -> list[list[LayoutElement]]

Execute a Cypher query and return results as layout elements.

For each row in the query results, model element nodes are resolved to their corresponding layout elements via the LayoutModelMapping.

Parameters:

Name	Type	Description	Default
query	str	The Cypher query string.	required
params	dict | None	Optional query parameters.	None

Returns:

Type	Description
list[list[LayoutElement]]	A list of rows, where each row is a list of LayoutElement objects.

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

def execute_query_as_layout_elements(
    self, query: str, params: dict | None = None
) -> list[list[momapy.core.elements.LayoutElement]]:
    """Execute a Cypher query and return results as layout elements.

    For each row in the query results, model element nodes are resolved
    to their corresponding layout elements via the LayoutModelMapping.

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

    Returns:
        A list of rows, where each row is a list of LayoutElement objects.
    """
    layout_element_results = []
    results = self.execute_query(query, params=params, resolve_nodes=True)
    layout_element_node_class = self._context.type_to_node_class.get(
        momapy.core.elements.LayoutElement
    )
    model_element_node_class = self._context.type_to_node_class.get(
        momapy.core.elements.ModelElement
    )
    for row in results:
        row_values = momapy_kb.utils.flatten_collection(list(row.values()))
        layout_element_nodes = []
        for element in row_values:
            if not isinstance(element, fieldz_kb.lpg.graph.BaseNode):
                continue
            if layout_element_node_class is not None and isinstance(
                element, layout_element_node_class
            ):
                layout_element_nodes.append(element)
            elif model_element_node_class is not None and isinstance(
                element, model_element_node_class
            ):
                layout_element_nodes += (
                    self.get_layout_element_nodes_from_model_element_node(element)
                )
        layout_elements = [
            fieldz_kb.lpg.core.make_object_from_node(self._context, node)
            for node in layout_element_nodes
        ]
        to_add = []
        for layout_element in layout_elements:
            for sub_layout_element in [
                layout_element
            ] + layout_element.descendants():
                if isinstance(sub_layout_element, momapy.core.layout.Arc):
                    for attr_name in ["source", "target"]:
                        attr_value = getattr(sub_layout_element, attr_name, None)
                        if isinstance(
                            attr_value,
                            momapy.core.elements.LayoutElement,
                        ):
                            to_add.append(attr_value)
        layout_elements += to_add
        layout_element_results.append(layout_elements)
    return layout_element_results

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 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 the query must return whatever needs seeding. 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.

Raises:

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

Source code in src/momapy_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 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 the query must
            return whatever needs seeding. 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.

    Raises:
        ValueError: If object_key_to_node is given and a converted object is
            not hashable.
    """
    return self._session.execute_query_as_objects(
        query,
        params=params,
        node_id_to_object=node_id_to_object,
        object_key_to_node=object_key_to_node,
    )

get_layout_element_nodes_from_model_element_node

get_layout_element_nodes_from_model_element_node(model_element_node: BaseNode) -> list[BaseNode]

Get layout element nodes linked to a model element node.

Queries the LayoutModelMapping in the database to find layout elements corresponding to the given model element.

Parameters:

Name	Type	Description	Default
model_element_node	BaseNode	A saved model element node.	required

Returns:

Type	Description
list[BaseNode]	A list of layout element nodes.

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

def get_layout_element_nodes_from_model_element_node(
    self, model_element_node: fieldz_kb.lpg.graph.BaseNode
) -> list[fieldz_kb.lpg.graph.BaseNode]:
    """Get layout element nodes linked to a model element node.

    Queries the LayoutModelMapping in the database to find layout
    elements corresponding to the given model element.

    Args:
        model_element_node: A saved model element node.

    Returns:
        A list of layout element nodes.
    """
    database_id_func_name = (
        self._session._pylpg_session._backend._database_id_func_name
    )
    query = f"""
        MATCH (mapping:LayoutModelMapping)-[:HAS_ITEM]->(item:Item)-[:HAS_VALUE]->(model_element:ModelElement),
        (item)-[:HAS_KEY]->(key)
        WHERE {database_id_func_name}(model_element) = $element_id
        RETURN key
    """
    results = self.execute_query(
        query,
        params={"element_id": model_element_node._database_id},
        resolve_nodes=True,
    )
    nodes = momapy_kb.utils.flatten_collection(
        [list(row.values()) for row in results]
    )
    layout_element_node_class = self._context.type_to_node_class.get(
        momapy.core.elements.LayoutElement
    )
    layout_element_nodes = []
    for node in nodes:
        if isinstance(node, fieldz_kb.lpg.graph.BaseNode):
            if layout_element_node_class is not None and isinstance(
                node, layout_element_node_class
            ):
                layout_element_nodes.append(node)
            else:
                layout_element_nodes += node.items.all()
    return layout_element_nodes

make_layout_elements_from_model_element_node

make_layout_elements_from_model_element_node(model_element_node: BaseNode) -> list[LayoutElement]

Get layout elements as Python objects for a model element node.

Parameters:

Name	Type	Description	Default
model_element_node	BaseNode	A saved model element node.	required

Returns:

Type	Description
list[LayoutElement]	A list of momapy LayoutElement objects.

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

def make_layout_elements_from_model_element_node(
    self, model_element_node: fieldz_kb.lpg.graph.BaseNode
) -> list[momapy.core.elements.LayoutElement]:
    """Get layout elements as Python objects for a model element node.

    Args:
        model_element_node: A saved model element node.

    Returns:
        A list of momapy LayoutElement objects.
    """
    layout_element_nodes = self.get_layout_element_nodes_from_model_element_node(
        model_element_node
    )
    layout_elements = [
        fieldz_kb.lpg.core.make_object_from_node(self._context, node)
        for node in layout_element_nodes
    ]
    return layout_elements

save_collections_from_entries

save_collections_from_entries(collection_names_and_entries: list[tuple[str, list[CollectionEntry]]], integration_mode: Literal['id', 'hash'] = 'id', delete_all: bool = False, with_membership_edges: bool = False, object_key_to_node: dict | None = None) -> None

Save collections from pre-built CollectionEntry objects.

Parameters:

Name	Type	Description	Default
collection_names_and_entries	list[tuple[str, list[CollectionEntry]]]	List of (name, entries) tuples.	required
delete_all	bool	If True, clear the database before saving.	False
with_membership_edges	bool	If True, emit HAS_MEMBER_MODEL_ELEMENT and HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to every contained element.	False
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, or a dict seeded by execute_query_as_objects to reuse the nodes of the entries' objects instead of creating duplicates.	None

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

def save_collections_from_entries(
    self,
    collection_names_and_entries: list[
        tuple[str, list[momapy_kb.core.CollectionEntry]]
    ],
    integration_mode: typing.Literal["id", "hash"] = "id",
    delete_all: bool = False,
    with_membership_edges: bool = False,
    object_key_to_node: dict | None = None,
) -> None:
    """Save collections from pre-built CollectionEntry objects.

    Args:
        collection_names_and_entries: List of (name, entries) tuples.
        delete_all: If True, clear the database before saving.
        with_membership_edges: If True, emit HAS_MEMBER_MODEL_ELEMENT and
            HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to
            every contained element.
        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, or a dict seeded by
            `execute_query_as_objects` to reuse the nodes of the entries'
            objects instead of creating duplicates.
    """
    if delete_all:
        self.delete_all()
    collections = []
    for (
        collection_name,
        collection_entries,
    ) in collection_names_and_entries:
        collection = momapy_kb.core.Collection(
            name=collection_name, entries=frozenset(collection_entries)
        )
        collections.append(collection)
    self.save_from_objects(
        collections,
        integration_mode=integration_mode,
        with_membership_edges=with_membership_edges,
        object_key_to_node=object_key_to_node,
    )

save_collections_from_file_paths

save_collections_from_file_paths(collection_names_and_file_paths: list[tuple[str, list]], return_type: Literal['map', 'model', 'layout'] = 'map', integration_mode: Literal['id', 'hash'] = 'id', delete_all: bool = False, with_membership_edges: bool = False, object_key_to_node: dict | None = None) -> None

Load maps from files and save them as named collections.

Parameters:

Name	Type	Description	Default
collection_names_and_file_paths	list[tuple[str, list]]	List of (name, file_paths) tuples.	required
return_type	Literal['map', 'model', 'layout']	What to extract from files ("map", "model", or "layout").	'map'
delete_all	bool	If True, clear the database before saving.	False
with_membership_edges	bool	If True, emit HAS_MEMBER_MODEL_ELEMENT and HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to every contained element.	False
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 the collections they save.	None

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

def save_collections_from_file_paths(
    self,
    collection_names_and_file_paths: list[tuple[str, list]],
    return_type: typing.Literal["map", "model", "layout"] = "map",
    integration_mode: typing.Literal["id", "hash"] = "id",
    delete_all: bool = False,
    with_membership_edges: bool = False,
    object_key_to_node: dict | None = None,
) -> None:
    """Load maps from files and save them as named collections.

    Args:
        collection_names_and_file_paths: List of (name, file_paths) tuples.
        return_type: What to extract from files ("map", "model", or "layout").
        delete_all: If True, clear the database before saving.
        with_membership_edges: If True, emit HAS_MEMBER_MODEL_ELEMENT and
            HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to
            every contained element.
        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 the collections they save.
    """
    if delete_all:
        self.delete_all()
    collection_names_and_entries = []
    for (
        collection_name,
        collection_file_paths,
    ) in collection_names_and_file_paths:
        collection_entries = []
        for file_path in collection_file_paths:
            reader_result = momapy.io.core.read(file_path, return_type=return_type)
            entry_id = file_path.stem
            collection_entry = momapy_kb.core.CollectionEntry(
                id_=entry_id,
                obj=reader_result.obj,
                file_path=str(file_path),
                element_to_annotations=reader_result.element_to_annotations,
                id_to_element=reader_result.id_to_element,
                source_id_to_model_element=reader_result.source_id_to_model_element,
                source_id_to_layout_element=reader_result.source_id_to_layout_element,
            )
            collection_entries.append(collection_entry)
        collection_names_and_entries.append(
            (
                collection_name,
                collection_entries,
            )
        )
    self.save_collections_from_entries(
        collection_names_and_entries,
        delete_all=delete_all,
        integration_mode=integration_mode,
        with_membership_edges=with_membership_edges,
        object_key_to_node=object_key_to_node,
    )

save_from_file

save_from_file(file_path: str, return_type: Literal['map', 'model', 'layout'] = 'map', with_layout: bool = True, with_model: bool = True, integration_mode: Literal['hash', 'id'] | None = None, with_membership_edges: bool = False, object_key_to_node: dict | None = None) -> None

Load a map from a file and save it to the database.

Supports CellDesigner, SBGN, and SBML file formats.

Parameters:

Name	Type	Description	Default
file_path	str	Path to the input file.	required
return_type	Literal['map', 'model', 'layout']	What to extract from the file ("map", "model", or "layout").	'map'
with_layout	bool	Whether to include layout data.	True
with_model	bool	Whether to include model data.	True
integration_mode	Literal['hash', 'id'] | None	How to handle duplicate objects ("hash" or "id").	None
with_membership_edges	bool	If True, emit HAS_MEMBER_MODEL_ELEMENT and HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to every contained element.	False
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 the files they load.	None

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

def save_from_file(
    self,
    file_path: str,
    return_type: typing.Literal["map", "model", "layout"] = "map",
    with_layout: bool = True,
    with_model: bool = True,
    integration_mode: typing.Literal["hash", "id"] | None = None,
    with_membership_edges: bool = False,
    object_key_to_node: dict | None = None,
) -> None:
    """Load a map from a file and save it to the database.

    Supports CellDesigner, SBGN, and SBML file formats.

    Args:
        file_path: Path to the input file.
        return_type: What to extract from the file ("map", "model", or "layout").
        with_layout: Whether to include layout data.
        with_model: Whether to include model data.
        integration_mode: How to handle duplicate objects ("hash" or "id").
        with_membership_edges: If True, emit HAS_MEMBER_MODEL_ELEMENT and
            HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to
            every contained element.
        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 the files they load.
    """
    object_ = momapy.io.core.read(
        file_path=file_path,
        return_type=return_type,
        with_model=with_model,
        with_layout=with_layout,
    ).obj
    self.save_from_object(
        object_,
        integration_mode=integration_mode,
        with_membership_edges=with_membership_edges,
        object_key_to_node=object_key_to_node,
    )

save_from_object

save_from_object(object_: object, integration_mode: Literal['hash', 'id'] = 'id', exclude_from_integration: tuple[type, ...] | None = None, with_membership_edges: bool = False, 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
with_membership_edges	bool	If True, emit HAS_MEMBER_MODEL_ELEMENT and HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to every contained element.	False
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, or a dict seeded by execute_query_as_objects to update the existing nodes instead of creating duplicates. 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/momapy_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,
    with_membership_edges: bool = False,
    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.
        with_membership_edges: If True, emit HAS_MEMBER_MODEL_ELEMENT and
            HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to
            every contained element.
        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, or a dict seeded by
            `execute_query_as_objects` to update the existing nodes instead
            of creating duplicates. 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(
        [object_],
        integration_mode=integration_mode,
        exclude_from_integration=exclude_from_integration,
        with_membership_edges=with_membership_edges,
        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, with_membership_edges: bool = False, 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
with_membership_edges	bool	If True, emit HAS_MEMBER_MODEL_ELEMENT and HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to every contained element. Only elements present in object_key_to_node get an edge: a cache shared with an earlier save already holds them, but a cache seeded from a query holds only the objects the query returned, and a cached root short-circuits the walk over its own descendants, so seed the elements too or leave this off for seeded saves.	False
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, or a dict seeded by execute_query_as_objects to update the existing nodes instead of creating duplicates. 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/momapy_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,
    with_membership_edges: bool = False,
    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.
        with_membership_edges: If True, emit HAS_MEMBER_MODEL_ELEMENT and
            HAS_MEMBER_LAYOUT_ELEMENT edges from each Model/Layout root to
            every contained element. Only elements present in object_key_to_node
            get an edge: a cache shared with an earlier save already holds
            them, but a cache seeded from a query holds only the objects
            the query returned, and a cached root short-circuits the walk
            over its own descendants, so seed the elements too or leave
            this off for seeded saves.
        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, or a dict seeded by
            `execute_query_as_objects` to update the existing nodes instead
            of creating duplicates. 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.
    """
    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
    if with_membership_edges:
        all_relationships += self._make_membership_relationships(
            objects, object_key_to_node, integration_mode
        )
    self._session._pylpg_session.save(all_nodes)
    self._session._pylpg_session.save(all_relationships)