eleanor.backend.service

Eleanor Framework Service Package

Services are the core of the Eleanor framework and are where user-accessible capabilities are implemented. The design guidelines for services are as follows:

Services may invoke other services, however the dependency should be documented.
Services may invoke DAOs for RDBMS access.
Services must NOT depend on any DTO objects from the API layer as services are API version agnostic.
Services may be long-running and can be invoked via the API layer, scheduled jobs, or the future unit-of-work scheduling feature, see #180.

Attributes

D `module-attribute`

D = TypeVar('D')

R `module-attribute`

R = TypeVar('R')

S `module-attribute`

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

Classes

Functions

async_run_service `async`

async_run_service(
    *,
    service_cls: Type[S],
    executor: Executor,
    method: Callable[..., R],
    service_kwargs: Optional[Dict[str, Any]] = None,
    method_kwargs: Optional[Dict[str, Any]] = None,
    dto_mapper: Optional[Callable[[R], D]] = None
) -> D

Asynchronously initializes a service instance with a new RDBMS and Neo4j session, executes a specified method using the provided executor, and optionally maps the result using a DTO mapper function.

This method is designed to handle cases where long-running or compute-intensive operations need to be executed in a separate thread while maintaining database session integrity. It ensures that each service instance has access to fresh RDBMS and Neo4j sessions, and manages the lifecycle of those sessions within the context of the background thread.

Rationale

FastAPI’s default session management is tied to the lifecycle of a single HTTP request. For long-running operations or tasks that involve database access outside the standard request-response cycle, this can lead to issues like expired sessions or connection timeouts.
async_run_service solves this problem by creating new RDBMS and Neo4j sessions within a background thread using persistence_sessions. This keeps database interactions contained within a scoped lifecycle, ensuring that connections are properly managed and cleaned up after use.
The method parameter allows for dynamically specifying which service method to execute, enabling flexibility for different tasks without needing to modify this function. It accepts a method of the service class, which is invoked with the provided method_kwargs.
The dto_mapper parameter allows for transforming the service method’s output while the database session is still active, which is crucial for avoiding issues with detached or expired SQLAlchemy ORM objects. Without this, trying to access related data after the session has been closed would raise errors.

Parameters:

service_cls (Type[S]) –

The service class to be instantiated. Must inherit from BaseService. This ensures consistency in how services interact with the database and enforces a common interface.
executor (Executor) –

The executor to be used for running the service method. Typically, this is a ThreadPoolExecutor to prevent blocking the FastAPI event loop during long-running operations.
method (Callable[..., R]) –

A method of the service class to be called. This should be a method that accepts the service instance as its first parameter and performs the desired action (e.g., fetching data from the database).
service_kwargs (Optional[Dict[str, Any]], default: None ) –

Optional keyword arguments to be passed to the service class constructor. These can include parameters specific to how the service should be configured or initialized.
method_kwargs (Optional[Dict[str, Any]], default: None ) –

Optional keyword arguments to be passed to the service method. These are used to specify parameters required by the service method (e.g., IDs for fetching data).
dto_mapper (Optional[Callable[[R], D]], default: None ) –

Optional function to transform the service method result into a DTO or other data structure while the database session is still active. This is particularly useful for converting ORM objects into a format that can be safely returned in an API response without encountering “detached object” errors.

Returns:

D –

The result of the DTO mapping function if provided, otherwise the raw service method result.
D –

The return type is dictated by the dto_mapper parameter, allowing flexibility
D –

in how the service output is transformed.

Raises:

Exception –

If any errors occur during the creation of sessions, method execution, or DTO transformation. These are propagated back to the caller.

Example:

```python
  response = await async_run_service(
      service_cls=MemoryCollectionService,
      executor=job_pool,
      method=MemoryCollectionService.get_memory_collection,
      method_kwargs={"memory_collection_pkid": collection_pkid},
      dto_mapper=lambda x: dto.CollectionGetResponseV1(
          collection=dto.CollectionV1.from_orm_model(x)
      ),
  )
```

run_background_service

run_background_service(
    *,
    service_cls: Type[S],
    executor: Executor,
    method: Callable[..., R],
    task_id: str,
    service_kwargs: Optional[Dict[str, Any]] = None,
    method_kwargs: Optional[Dict[str, Any]] = None
) -> None

Run a service method in the background using the specified executor.

This method manages the lifecycle of RDBMS and Neo4j sessions for background tasks, ensuring that sessions are properly scoped within the thread running the background task. It logs the start and end of each task, as well as any exceptions that occur.