edri.abstract.manager package

Submodules

edri.abstract.manager.async_manager_base module

class edri.abstract.manager.async_manager_base.AsyncManagerBase(router_queue: Any | None = None, *args: Any, **kwargs: Any)

Bases: ManagerBase

A ManagerBase whose event loop is asyncio-driven instead of a blocking multiprocessing.connection.wait() loop.

Everything else — the metaclass's @request/@response registration, _collect_vitals/health_check, start_worker/WorkerProcess, run()'s startup sequence — is inherited from ManagerBase unchanged; this class only overrides how incoming events are waited for and dispatched.

A @request/@response handler may now be async def. Each async handler is dispatched as its own asyncio.Task: the loop does not wait for it to finish before reading the next pipe, so a handler that needs to do I/O can simply await it instead of calling start_worker with a WorkerThread just to avoid blocking the loop. Plain def handlers keep working exactly as they do on ManagerBase today — called inline, synchronously, in the order their pipe becomes readable — so existing synchronous handlers need no code change.

WorkerProcess (CPU-bound work needing real parallelism) is untouched: asyncio does not grant extra cores, so CPU-bound work still belongs in a WorkerProcess, offloaded via start_worker exactly as before.

Async handlers reuse ManagerBase's cache, bookkeeping, response preparation, and router-output stages. This keeps their behavior aligned with synchronous handlers, including calls to prepare_outgoing_event.

Pipe reads deliberately remain synchronous once the event loop reports a descriptor readable. That mirrors ManagerBase.wait() -> Connection.recv() and, critically, preserves multiprocessing.Connection's blocking framed protocol. uvloop/libuv changes fds passed to add_reader() to O_NONBLOCK; this class restores blocking mode immediately after every reader registration.

resolve(event: Event) → None

Resolves one event through independently overridable stages.

run_resolver() → None

Runs resolver cycles until interrupted or explicitly stopped.

edri.abstract.manager.manager_base module

class edri.abstract.manager.manager_base.ManagerBase(router_queue: Any | None = None, logger: Logger | None = None, from_time: datetime | None = None, max_workers: int = -1)

Bases: ABC, Process

Base class for a manager that handles events and manages worker processes or threads.

The manager subscribes to request and response event types, resolves incoming events, and may delegate work to worker threads or processes. A configurable worker limit can be applied through max_workers. When the limit is reached, additional worker start requests are queued and launched later as running workers finish.

Attributes:

router_queue (Optional["Queue[Event]"]):

Queue for communicating with the router.

logger (Optional[Logger]):

Logger instance for logging events.

_from_time (Optional[datetime]):

Start time for filtering events.

max_workers (int):

Maximum number of workers allowed to run concurrently. -1 means unlimited and is the default.

Methods:

__init__: Constructor for initializing the manager. _find_worker: Finds a worker based on its communication pipe. _prepare: Prepares the manager before starting the event resolver loop. _prepare_resolvers: Prepares methods for resolving request and response events. _remove_worker: Removes a worker from the manager's tracking. _resolve_undelivered_events: Resolves any undelivered events. _subscribe: Subscribes to events based on requests and responses. _subscribe_static_resolvers: Subscribes to static event resolvers. additional_pipes: Property for specifying additional pipes. name: Property for specifying the manager's name. workers: Property for defining workers and their associated events. after_start: Hook for actions after the manager starts. get_pipes: Gets a set of all pipes including the router and worker pipes. health_check: Handles health check events from the router. resolve: Resolves an event through independently overridable stages. prepare_outgoing_event: Hook run immediately before router queue output. send_to_router: Common output path for every event sent to the router. resolve_callback_pipe: Handles events received from unknown pipes. resolve_callback_worker: Default method for resolving events received from workers. resolve_unknown: Handles unknown events received from the router pipe. run: Main entry point for the manager process. run_resolver: Main loop for resolving events from all pipes. run_resolver_cycle: Executes one independently testable resolver iteration. receive_pipe_event: Applies the shared receive and pipe-error policy. start: Starts the manager process with an optional router queue. start_worker: Starts a new worker for handling an event. store_get: Handles store get requests.

property additional_pipes: set[Connection]

Specifies additional pipes for the manager to listen on. Override to add custom pipes.

Returns:

A set of additional pipes.

Return type:

set[Connection]

after_start() → None

Hook method called after the manager process starts. Can be overridden to perform initialization tasks.

close_failed_pipe(pipe: Connection) → None

Closes a failed pipe while suppressing secondary close errors.

collect_custom_vitals() → dict[str, Any]

Add application-defined metrics to the health-check vitals — override hook.

The base implementation reports nothing. Override to attach your own gauges and counters, which ride along on the periodic health round-trip and appear under vitals.custom on the status endpoint:

def collect_custom_vitals(self) -> dict[str, Any]:
    return {
        "queue_depth": self._inbox.qsize(),
        "cache_entries": len(self.cache),
        "last_sync_age_seconds": time() - self._last_sync,
    }

Keep it cheap and non-blocking: it runs inline on the component's main loop on every check. Return JSON-serializable primitives so the values survive both the pipe round-trip and the endpoint's JSON encoding. Any exception raised here is caught and logged — it degrades to empty custom vitals rather than wiping out the standard ones.

Returns:

Custom metric name → value. Empty by default.

Return type:

dict[str, Any]

disable_additional_pipe(pipe: Connection) → None

Excludes a failed pipe from future wait sets and closes it.

execute_event_resolver(event: Event, resolver: Callable[[...], Any]) → None

Executes the selected resolver.

get_event_resolver(event: Event) → Callable[[...], Any] | None

Returns the registered request or response resolver for event.

get_pipes() → set[Connection]

Collects all communication pipes used by the manager, including the router pipe and any worker pipes.

Returns:

A set of all pipes for the manager to listen on.

Return type:

set[Connection]

get_resolver_cache_state(event: Event, resolver: Callable[[...], Any]) → ResolverCacheState

Builds the cache state used while resolving one event.

handle_additional_pipe_receive_error(pipe: Connection, error: EOFError | OSError) → None

Disables an application-defined pipe after a failed read.

handle_pipe_receive_error(pipe: Connection, error: EOFError | OSError) → None

Dispatches a pipe read failure according to the pipe's owner.

handle_resolver_exception(event: Event, error: Exception) → None

Logs and stores an exception raised while resolving a router event.

handle_router_pipe_receive_error(pipe: Connection, error: EOFError | OSError) → None

Stops the resolver when its router connection is no longer usable.

handle_worker_pipe_receive_error(pipe: Connection, worker_key: str, worker: Worker, error: EOFError | OSError) → None

Removes a worker whose communication pipe has closed or failed.

health() → HealthReport

Assess this component's health for a health-check response — override hook.

The default derives status purely from exceptions caught since the last check (any → WARNING) and always reports the component as ready. Override to add domain knowledge — typically checking the dependencies this component needs and downgrading accordingly:

def health(self) -> HealthReport:
    report = super().health()
    if not self._database.is_connected():
        return HealthReport(Status.DEGRADED, ready=False,
                            reason="database unreachable")
    return report

The caught exceptions are still reported separately (for the detail view) regardless of what this returns, so an override should escalate status, not suppress exceptions. Keep it cheap and non-blocking: it runs on every health-check round-trip, inline on the component's main loop.

Returns:

The component's self-assessment (status, readiness, reason).

Return type:

HealthReport

health_check(event: HealthCheck) → None

Responds to a health check request from the router, indicating the manager's status.

Parameters:

event (HealthCheck) -- The health check event received from the router.

is_not_modified_request(event: Event, cache: ResolverCacheState) → bool

Returns whether event can be answered with 304 Not Modified.

is_router_pipe(pipe: Connection) → bool

Returns whether pipe is the manager's router connection.

property max_workers: int

property name: str

Specifies the manager's name. Override to provide a custom name.

Returns:

The manager's name.

Return type:

str

on_resolver_idle() → None

Handles a resolver cycle in which there are no pipes to monitor.

prepare_cache_response(event: Event, cache: ResolverCacheState) → None

Renews cache state or attaches HTTP cache response directives.

prepare_outgoing_event(event: Event) → None

Hook called immediately before an event enters the router queue.

prepare_resolved_event(event: Event, cache: ResolverCacheState) → None

Applies framework response metadata before the common outgoing hook.

quit() → None

receive_pipe_event(pipe: Connection) → Event | None

Receives one event using the shared pipe error policy.

record_resolved_event(event: Event) → None

Updates resolver health counters after successful resolver execution.

resolve(event: Event) → None

Resolves one event through independently overridable stages.

resolve_active_pipe(pipe: Connection) → None

Dispatches a readable pipe to its source-specific resolver.

resolve_callback_pipe(event: Event, pipe: Connection) → None

Handles events received from unknown pipes, which are not associated with known workers.

Parameters:

• event (Event) -- The received event.

• pipe (Connection) -- The pipe from which the event was received.

resolve_callback_worker(event: Event, worker: Worker) → None

Default method for handling events received from workers. It forwards received events to the router queue.

Parameters:

• event (Event) -- The received event.

• worker (Worker) -- Information about the worker that sent the event.

resolve_non_router_event(event: Event, pipe: Connection) → None

Routes an event from a worker pipe or an application-defined pipe.

resolve_non_router_pipe(pipe: Connection) → None

Receives and dispatches one event from a worker or additional pipe.

resolve_not_modified(event: Event, cache: ResolverCacheState) → bool

Sends a cached 304 Not Modified response when applicable.

resolve_router_event(event: Event) → None

Resolves one router event and records resolver failures.

resolve_router_pipe(pipe: Connection) → None

Receives and resolves one event from the router connection.

resolve_unknown(event: Event) → None

Handles unknown events received from the router pipe that do not match any registered request or response.

Parameters:

event (Event) -- The unknown event received.

resolve_worker_control_event(event: Manager, worker_key: str, worker: Worker) → None

Handles lifecycle and other control events received from a worker.

resolve_worker_event(event: Event, worker_key: str, worker: Worker) → None

Routes a worker event either as control traffic or application traffic.

resolver_idle_delay: float = 1.0

resolver_wait_timeout: float = 1.0

run() → None

Main entry point for the manager process. Sets up event subscriptions, prepares the manager, and enters the main event resolution loop.

run_resolver() → None

Runs resolver cycles until interrupted or explicitly stopped.

run_resolver_cycle() → None

Executes one iteration of the resolver loop.

send_to_router(event: Event) → None

Runs the common outgoing hook and sends event to the router.

should_send_resolved_event(event: Event, had_response: bool) → bool

Returns whether resolver execution created a changed response to send.

start(router_queue: Any | None = None) → None

Starts the manager process with an optional router queue specified. This method initiates the manager's operation, allowing it to begin processing events. The router queue is a crucial component for communication between the manager and the router, facilitating the dispatch and handling of events.

Parameters:

router_queue (Optional[Queue[Event]]) -- The event queue used for communication with the router. If provided, this queue will be set as the current router queue for the manager. This queue is essential for the manager to receive and process events.

Raises:

RuntimeError -- Raised if no router queue is provided and no router queue has been set previously. This exception ensures that the manager cannot start without a means to communicate with the router, as the lack of a router queue would prevent the manager from functioning correctly.

Note

If a router queue is not provided as an argument to this method, the manager will attempt to use a previously set router queue. It's essential to either provide a router queue when calling this method or set one before its execution. Failure to do so will result in a RuntimeError, indicating the manager's inability to start due to the absence of a communication channel with the router.

Example

To start the manager, you may either provide a router queue directly when calling the start method or ensure that the manager has a router queue set beforehand. This flexibility allows for different use cases, such as reusing an existing queue or setting a new one as needed.

start_worker(event: Event | None, resolver: WorkerFactoryProtocol, /, *args: Any, environment: Path | None = None, **kwargs: Any) → None

stop_resolver() → None

Requests a clean exit from run_resolver().

store_get(event: Get) → None

Handles the store get response for retrieving data from the Store. This method determines whether the response should be directed to a specific worker or handled by the manager itself.

The method first checks if the event is associated with a worker. If the event's _worker attribute is set and matches a worker managed by this instance, the event is sent to that worker via its communication pipe. If no matching worker is found, or if the _worker attribute is not set, the method attempts to handle the event using the _store_get method, if it is defined.

Parameters:

event (Get) -- The get response event, containing details of the data to retrieve and the retrieved data.

Behavior:

• If the event specifies a worker (via the _worker attribute), the method directs the response to that worker.

• If no worker is specified or found, the event is passed to the manager's own _store_get method for standard handling, provided that _store_get has been defined. If _store_get is not defined, the event is effectively ignored.

Note

• The _store_get method is an optional user-defined method. If it is not defined, this method will not attempt to handle the event beyond checking for a worker.

wait_for_active_pipes(pipes: set[Connection]) → list[Connection]

Waits until one or more monitored pipes become readable.

edri.abstract.manager.manager_priority_base module

class edri.abstract.manager.manager_priority_base.ManagerPriorityBase(router_queue: Any | None = None, logger: Logger | None = None, from_time: datetime | None = None, max_workers: int = -1)

Bases: ManagerBase

An extension of ManagerBase that introduces priority handling for events. This class manages events using a priority queue, ensuring that higher-priority events are processed before lower-priority ones.

Attributes:

• event_queue (PriorityQueue[QueueItem]) -- A priority queue for managing events based on their priority.

• event_worker (Thread) -- A thread dedicated to handling command execution based on event priority.

• waking_up (tuple[Connection, Connection]) -- A pair of connections used to wake up the manager when a new worker starts.

Methods:

• __init__ -- Initializes the ManagerPriorityBase with an optional router queue and logger.

• _prepare -- Prepares the manager by initializing the event queue, event worker thread, and waking up pipe.

• additional_pipes -- Property to include the waking_up pipe in the set of additional pipes the manager listens to.

• command_handler -- Thread target function that processes events from the priority queue.

• get_priority -- Abstract method that must be implemented to define the priority of events.

• run_resolver -- Main event loop, enhanced to handle events based on priority.

• start_worker -- Overrides the start_worker method to wake up the manager when a new worker is started.

property additional_pipes: set[Connection]

Includes the waking_up pipe in the set of additional pipes the manager listens to.

Returns:

A set of additional pipes, including the waking_up pipe.

Return type:

set[Connection]

command_handler() → None

The thread target function that processes events from the priority queue, ensuring that events are handled in order of their priority.

abstractmethod get_priority(event_type: Type[Event]) → int

Abstract method that determines the priority of an event. Must be implemented by subclasses to define how event priorities are assigned.

Parameters:

event_type (Type[Event]) -- The class of the event for which the priority needs to be determined.

Returns:

The priority of the event.

Return type:

int

run_resolver() → None

The main event loop for the manager, enhanced to handle events based on priority. It listens on all pipes, including the waking_up pipe, to efficiently process and prioritize events.

edri.abstract.manager.worker module

class edri.abstract.manager.worker.Worker(event: 'Event | None', resolver: 'WorkerFactoryProtocol', args: 'tuple[Any, ...]'=(), kwargs: 'dict[str, Any]'=<factory>, environment: 'Path | None' = None, policy: 'WorkerPolicy' = <factory>, state: 'WorkerState' = <WorkerState.PENDING: 1>, pipe: 'Connection | None' = None, runtime: 'WorkerRuntimeProtocol | None' = None, created: 'float' = 0.0, spawned: 'float | None' = None, started: 'float | None' = None, stop_requested: 'float | None' = None, finished: 'float | None' = None, stop_reason: 'WorkerStopReason | None' = None, timeout: 'WorkerTimeout | None' = None, outcome: 'WorkerOutcome | None' = None, exit: 'WorkerExit | None' = None, pending_responses: 'set[UUID]' = <factory>)

Bases: object

class edri.abstract.manager.worker.WorkerFactoryProtocol(*args, **kwargs)

Bases: Protocol

Constructor contract for thread and process worker classes.

exception edri.abstract.manager.worker.WorkerNotFoundError

Bases: KeyError

Raised when cancellation targets a worker that is no longer registered.

class edri.abstract.manager.worker.WorkerRuntimeProtocol(*args, **kwargs)

Bases: Protocol

Runtime operations owned by WorkerController.

is_alive() → bool

start() → None

edri.abstract.manager.worker_controller module

class edri.abstract.manager.worker_controller.WorkerController(manager: ManagerBase, policy: WorkerPolicy | None = None)

Bases: Mapping[str, Worker]

Owns a Manager's worker registry, scheduling, lifecycle and deadlines.

cancel(worker_id: str, reason: WorkerStopReason = WorkerStopReason.REQUESTED) → None

cancel_all() → int

find_by_pipe(pipe: Any) → tuple[str, Worker]

finish_from_quit(worker_id: str, event: WorkerQuit) → None

finish_lost(worker_id: str) → None

property pending: WorkerView

pipes() → set[Any]

property running: WorkerView

start(event: Event | None, resolver: WorkerFactoryProtocol, /, *args: Any, environment: Path | None = None, policy: WorkerPolicy | None = None, **kwargs: Any) → str

started(worker_id: str, at: float | None = None) → None

property stopping: WorkerView

class edri.abstract.manager.worker_controller.WorkerView(controller: WorkerController, states: frozenset[WorkerState])

Bases: Mapping[str, Worker]

A stable, live, read-only view over selected controller states.

Module contents