persistence

Persistence Context Managers

This module provides context managers for managing persistence connections to the backend API (currently RDBMS and Memgraph).

Attributes

T module-attribute

T = TypeVar('T', bound=BaseService)

Classes

Functions

persistence_sessions

persistence_sessions(
    **neo4j_kwargs,
) -> Generator[Tuple[SASession, NJSession], None, None]

Creates both RDBMS and Neo4j sessions and yields them as a context manager.

This context manager allows the creation of both sessions using a single with statement. The sessions are automatically committed if no exceptions occur during the execution of the code block. If an exception occurs, the sessions are rolled back (for RDBMS) and closed. Finally, both sessions are closed.

Example:

```python
with persistence_sessions() as (sa_session, neo4j_session):
    # Code block using both sessions...
```

Parameters:

• **neo4j_kwargs –

  Additional keyword arguments to pass to the Neo4j session creation.

Yields:

• Tuple[SASession, NJSession] –

  Tuple[SASession, NJSession]: A tuple containing the RDBMS and Neo4j sessions.

Raises:

• Exception –

  If an exception occurs during the execution of the code block.

service_context

service_context(
    *service_classes: Type[T], **neo4j_kwargs
) -> Generator[Tuple[T, ...], None, None]

Creates instances of the given service classes with pre-initialized RDBMS and Neo4j session instances.

Example:

Using a single service:

```python
with service_context(MyServiceClass) as (service,):
    # Use `service` here...
    service.perform_action()
```

Using multiple services:

```python
with service_context(
        AgentService, MemoryCollectionService, NamespaceService
    ) as (
        agent_service,
        collection_service,
        namespace_service,
    ):
        agent_service = cast(AgentService, agent_service)
        collection_service = cast(MemoryCollectionService, collection_service)
        namespace_service = cast(NamespaceService, namespace_service)
```

Parameters:

• *service_classes (Type[T], default: () ) –

  One or more service classes to instantiate, implementing BaseService.

• **neo4j_kwargs –

  Additional keyword arguments to pass to the Neo4j session creation.

Yields:

• Tuple[T, ...] –

  Tuple[T, …]: A tuple containing instances of the service classes.

Raises:

• Exception –

  If an exception occurs during the execution of the code block.