Manager¶

A Manager is a core component responsible for handling Event instances delivered by the Router and executing the required application logic.

Managers are single-threaded by default and act as deterministic event processors. Concurrency is achieved explicitly via Workers.

There are three base classes:

ManagerBase – FIFO processing
ManagerPriorityBase – Priority FIFO processing (extended behavior)
AsyncManagerBase – Experimental: asyncio-driven event loop variant that supports async def handlers

Responsibilities¶

A Manager is responsible for:

Subscribing to events
Handling incoming events
Producing responses (if applicable)
Emitting new events when needed
Delegating heavy work to Workers

Managers do not route events. They receive events from the Router and may emit new events back to the Router via self.router_queue.put(...).

Event Subscription¶

Managers subscribe to events using decorators:

@request
@response

At startup, the framework scans decorated methods and registers the Manager as a subscriber for the corresponding event types.

Example¶

from edri.abstract import ManagerBase, request
from edri.dataclass.response import ResponseStatus
from my_events import Ping

class ExampleManager(ManagerBase):

    @request
    def ping(self, event: Ping) -> None:
        event.response.message = "pong"

If any field in event.response is set, the response status is automatically set to OK.

If event.response.set_status(...) is called explicitly, the status is set immediately, but the event is still sent only after the handler function finishes execution.

This design intentionally prevents developers from running background work in the manager thread after finishing event handling — once the handler returns, the event lifecycle continues.

Handling Events¶

When a Manager receives an event:

The correct decorated handler method is selected
The method is executed synchronously
If the event has a response and it is modified, the response is sent automatically

To emit a new event manually:

self.router_queue.put(new_event)

Managers must explicitly emit new events using self.router_queue.put(...).

New events are never sent automatically — only responses of the currently handled event are sent automatically at the end of the handler.

Unhandled events (`resolve_unknown`)¶

If an event arrives from the Router that matches no @request or @response handler on this Manager, it is passed to resolve_unknown:

def resolve_unknown(self, event: Event) -> None:
    # default: log a warning
    ...

The default implementation only logs a warning, so unrecognized events are otherwise dropped. Override it to provide a catch-all — for example to route events handled dynamically rather than by decorator, or to raise instead of silently ignoring them.

This is the fallback for events on the router pipe. Events arriving on a custom pipe instead go to resolve_callback_pipe, and a Worker's events go to resolve_callback_worker.

Concurrency with Workers¶

Managers remain single-threaded. For parallel or long-running work, use Workers.

To start a worker:

self.workers.start(event, WorkerClass)

Workers can be:

WorkerThread
WorkerProcess

Receiving events back from a Worker (`resolve_callback_worker`)¶

Every event a Worker emits with self.event_send(...) arrives at the Manager through a single callback:

def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
    ...

It is called once per event, for every worker — there is no other path by which a Worker's output enters the system.

What the default does. Out of the box, resolve_callback_worker tags the event with the worker's name (event._worker), records a timing stamp, and puts the event on the router queue. From there the Router dispatches it by type and direction — typically to a @response handler (in this or another Manager) or back to the waiting API client.

A Worker's event always travels Worker → callback → Router first; it never lands directly in a @response handler. The Manager is the only mediator of Worker traffic, which is exactly what makes this callback a useful interception point.

The worker argument. Alongside the event you receive the originating Worker, which lets you correlate the result with the task that produced it:

worker.event — the event that started this worker (the original request)
worker.runtime.name — the worker runtime's generated name (once spawned)

When to override. Override it to post-process worker output before it leaves the Manager — for example to aggregate several streamed results into one, tag or enrich events, or drop/filter some of them:

def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
    if isinstance(event, PartialResult):
        self._buffer.append(event)          # collect, emit nothing yet
        return
    super().resolve_callback_worker(event, worker)   # default: forward to the Router

Events arriving on a custom pipe (not from a Worker) go to a different callback, resolve_callback_pipe — see Listening on custom pipes.

Outgoing-event hook: `prepare_outgoing_event()`¶

Override prepare_outgoing_event(event) to inspect or modify an event immediately before a Manager places it on the Router queue:

from edri.abstract import ManagerBase
from edri.dataclass.event import Event


class AuditedManager(ManagerBase):
    def prepare_outgoing_event(self, event: Event) -> None:
        self.logger.info("Sending %s", type(event).__name__)

The hook is called for every Router-bound event emitted by ManagerBase or its framework-provided subclasses. This includes resolved responses, Worker callback output, scheduled events, Store callbacks, API Broker output, and framework subscription events. AsyncManagerBase applies the same hook to both synchronous and asynchronous handler output.

For resolved responses, framework response preparation happens first, so the hook sees the final event state, including cache directives and updated Switch receive state. Any changes made by the hook are visible to the Router.

Keep the hook fast and non-blocking because it runs inline with the Manager's event processing. If the hook needs event-specific fields, check the event type first: it also receives framework control events such as subscriptions. Override this hook instead of send_to_router() so the framework retains a single, consistent queue-output path.

Lifecycle hook: `after_start()`¶

Managers provide an after_start() hook that runs once, after the manager process starts.

Use it to perform initialization tasks that must happen inside the manager process (for example: warming caches, preparing in-memory state, seeding timestamps, opening long-lived resources, etc.).

def after_start(self) -> None:
    """Hook method called after the manager process starts. Can be overridden to perform initialization tasks."""

Reporting health¶

Every Manager answers the Router's periodic HealthCheck automatically. Two hooks let a Manager enrich what it reports:

health() — return a HealthReport(status, ready, reason) to self-assess health and readiness (e.g. downgrade to DEGRADED when a dependency is down).
collect_custom_vitals() — return a dict[str, Any] of application-defined metrics that ride along on the same round-trip and show up under vitals.custom.

Both run inline on the main loop on every check, so keep them cheap. See Health checking for the full contract, the status endpoint, and the vitals reference.

Message Processing Strategies¶

Managers process incoming events in one of two modes:

FIFO (`ManagerBase`)¶

Events are processed in arrival order.

Priority FIFO (`ManagerPriorityBase`)¶

Events are processed by priority (lower value = higher priority). Events with equal priority follow FIFO order.

Detailed priority configuration is described separately.

Manager-Level Caching¶

Managers support HTTP cache integration via the @request decorator.

Example:

@request(cache="information")
def get_information(self, event: InformationEvent) -> None:
    ...

How it works¶

The cache argument defines a cache key namespace.
When used with a GET HTTP request, EDRI integrates with HTTP caching semantics.

ETag handling¶

To support ETag-based caching, the event must expose the If-None-Match header as a field.

Example:

etag: list[str] | None = Header(name="If-None-Match")

An ETag is generated in the format:
```
<cache-key>-<last-change-timestamp-in-microseconds>
```
The timestamp carries microsecond resolution, so two changes within the same second produce distinct ETags.

This cache ETag (identifier + timestamp, validated via If-None-Match) is a separate mechanism from the file-download ETag, which is a content fingerprint validated via If-Range. See Two unrelated ETag mechanisms.
If the incoming request contains If-None-Match (provided via the etag field) and it matches the current ETag, the response status is set to 304 Not Modified.
If it does not match:
- The regular response is returned.
- The following HeaderResponseDirectives are added automatically:
  - ETag
  - Cache-Control
  - Vary

Cache-Control¶

Cache-Control is controlled by the environment variable:

EDRI_API_CACHE_CONTROL

Vary header¶

By default, Vary contains:

Accept,Accept-Encoding

Additional headers can be appended using:

EDRI_API_CACHE_HEADERS

Cache invalidation behavior¶

Automatic: a non-GET request (POST, PUT, PATCH, DELETE) whose handler carries the same cache="..." key and returns an OK response bumps the "last change" timestamp for that key. The next GET for that key then produces a new ETag and returns fresh data.

Manual: the resolver cache is exposed as self.cache on the Manager. Call renew() to invalidate a key from anywhere — a handler that does not share the cache="..." key, a worker callback, or in response to an external event:

@response
def on_upstream_changed(self, event: UpstreamChanged) -> None:
    self.cache.renew("information")

Cache: setting an initial timestamp¶

To pre-seed a cache key's timestamp at startup, override after_start() and call seed():

def after_start(self) -> None:
    super().after_start()
    self.cache.seed("information", datetime(2024, 1, 1, tzinfo=timezone.utc))

Caching is automatic once cache="..." is defined on the handler.

self._cache is a deprecated alias for self.cache and will be removed in a future release.

Key-Value Store (Specialized Manager)¶

EDRI provides a built-in Key-Value Store implemented as a specialized Manager.

It enables:

Data caching
Cross-manager data sharing
Temporary storage of computed results

It is started explicitly via:

edri.start_store()

Full details are described in a dedicated document.

Scheduler (Specialized Manager)¶

The Scheduler is a time-based event emitter implemented as a specialized Manager.

It can:

Emit events at specific times
Repeat events at defined intervals

It is started via:

edri.start_scheduler()

Full scheduling behavior is documented separately.

Priority-based processing (`ManagerPriorityBase`)¶

ManagerPriorityBase is an extension of ManagerBase that processes events in priority order rather than strict FIFO.

When to use¶

Use ManagerPriorityBase when some event types must be handled before others regardless of arrival order — for example, cancellation signals should preempt normal work, or health-check events must always respond promptly.

How to subclass¶

You must implement the abstract get_priority(event_type) method. It returns an integer — lower value = higher priority (processed first).

ManagerPriorityBase.__init__ accepts the same parameters as ManagerBase:

OrderManager(
    router_queue=edri.router_queue,
    logger=None,          # optional custom Logger
    from_time=None,       # optional datetime — ignore events older than this
    max_workers=-1,       # optional concurrency limit (-1 = unlimited)
)

from typing import Type
from edri.abstract import ManagerPriorityBase, request
from edri.dataclass.event import Event
from events.orders import UrgentOrder, NormalOrder, LowPriorityReport


class OrderManager(ManagerPriorityBase):

    def get_priority(self, event_type: Type[Event]) -> int:
        priority_map = {
            UrgentOrder: 1,
            NormalOrder: 5,
            LowPriorityReport: 10,
        }
        return priority_map.get(event_type, 5)

    @request
    def handle_urgent(self, event: UrgentOrder) -> None:
        event.response.status = "dispatched"

    @request
    def handle_normal(self, event: NormalOrder) -> None:
        event.response.status = "queued"

    @request
    def handle_report(self, event: LowPriorityReport) -> None:
        event.response.generated = True

How it works internally¶

ManagerPriorityBase introduces a second internal thread (command_handler) that reads from a PriorityQueue[QueueItem]. The main event loop moves every incoming event into that priority queue. The command_handler thread then processes events in priority order.

        flowchart TD
    In["Incoming events from Router"] --> PQ["PriorityQueue<br/>(sorted by get_priority())"]
    PQ --> CH["command_handler thread"]
    CH --> Res["resolve(event) → @request handler"]

Events with equal priority are processed in FIFO order among themselves.

Caveats¶

ManagerPriorityBase uses an additional background thread. Keep get_priority() fast — it is called on every incoming event in the main loop.
Worker runtimes created via workers.start(...) send a wake signal to the priority loop so that worker events do not block the queue. This happens automatically.
All other ManagerBase features (after_start(), caching, @response, Workers) work identically.

Worker concurrency limit (`max_workers`)¶

Both ManagerBase and ManagerPriorityBase accept a max_workers parameter that limits how many workers can run simultaneously.

class MyManager(ManagerBase):
    def __init__(self, router_queue):
        super().__init__(router_queue, max_workers=4)

Default is -1 (unlimited).
When the limit is reached, additional workers.start(...) records remain PENDING in the controller's ordered storage.
As running workers finish (WorkerQuit), queued worker starts are drained automatically.
Setting max_workers = 0 or any value below -1 raises ValueError.

Listening on custom pipes (`additional_pipes`)¶

A Manager's event loop normally waits on two kinds of pipes: the router pipe and its Workers' pipes. To make it also react to an external multiprocessing.connection.Connection — a hardware device, a socket bridge, a background thread, another process — override the additional_pipes property:

from multiprocessing.connection import Connection

class DeviceManager(ManagerBase):
    @property
    def additional_pipes(self) -> set[Connection]:
        return {self._device_pipe}

    def resolve_callback_pipe(self, event, pipe: Connection) -> None:
        # Called for every object read off a pipe that is neither the
        # router pipe nor a known Worker pipe.
        self.handle_device_message(event)

The property is re-read on every loop iteration, so the set may change over time. Two things to keep in mind:

You must also override resolve_callback_pipe. Objects arriving on a custom pipe are not Worker events, so the loop routes them to resolve_callback_pipe(obj, pipe) — whose default implementation only logs a warning. Without an override the messages are effectively dropped.
Anything read off these pipes is whatever the other end sent; it does not have to be an Event. The dispatch by @request / @response does not apply here.

This is the same extension point ManagerPriorityBase uses internally (for its wake-up pipe) and that the Broker uses to listen to the Listener.

Execution Model Summary¶

Managers are single-threaded event processors
Subscriptions are declared via decorators
Responses are auto-sent when modified
New events must be emitted manually
Concurrency is achieved via Workers
Priority handling is available through ManagerPriorityBase

Managers are deterministic coordination units in the EDRI architecture — Workers perform the heavy lifting.