edri package

Subpackages

Submodules

edri.testing module

Testing helpers for EDRI application tests.

This module provides lightweight helpers for writing unit tests against Manager handlers without launching child processes. It wraps internal setup steps so application tests can use public API only.

ManagerTestHarness is Experimental — it may be extended in minor releases but breaking changes will follow the project's compatibility policy.

class edri.testing.ManagerTestHarness(manager: ManagerBase)[source]

Bases: object

Wraps a Manager instance for unit testing.

Usage:

harness = ManagerTestHarness(MyManager(router_queue=mock_queue))
harness.prepare()

event = MyEvent(...)
harness.resolve(event)

assert harness.get_status(event) == ResponseStatus.OK
assert harness.get_directives(event) == []

All assertions use public API so tests survive internal refactoring.

static get_directives(event: Event) → list[ResponseDirective][source]

Return the directives attached to event's response.

static get_status(event: Event) → ResponseStatus[source]

Return the response status after resolving event.

property manager: ManagerBase
prepare() → None[source]

Set up resolver maps so that resolve() dispatches correctly.

Must be called before the first resolve() call (equivalent to what _prepare_resolvers() does internally).

resolve(event: Event) → None[source]

Dispatch event through the manager's handler.

The manager's router_queue should be a mock for unit tests; real queues are only needed for integration tests.

Module contents

class edri.EDRI[source]

Bases: object

The EDRI class serves as the main orchestrator for the system, initializing and managing the core components such as the router, scheduler, store, API, and connector. It provides methods to add components, start various services, and run the router.

Attributes:

  • router_queue (multiprocessing.Queue[Event]) -- The queue used for routing events between components.

  • components (set[edri.abstract.ManagerBase]) -- A set of manager components that are part of the system.

  • store (edri.utility.manager.Store) -- The store component for data storage.

  • scheduler (edri.utility.manager.scheduler.Scheduler) -- The scheduler component for scheduling tasks.

  • api (edri.api.listener.Listener) -- The API listener component.

  • grpc (edri.api.grpc.listener.GRPCListener) -- The gRPC API listener component.

  • broker (edri.api.broker.Broker) -- The broker component for handling middleware.

  • connector (edri.router.connector.Connector) -- The connector component for connecting to external services.

  • router (edri.router.Router) -- The router component that routes events.

add_component(component: ManagerBase) → None[source]

Adds a component to the EDRI system.

Parameters:

component (edri.abstract.ManagerBase) -- The component to be added.

run() → None[source]

Runs the router and all added components, handling graceful shutdown on SIGINT (KeyboardInterrupt) and SIGTERM.

SIGTERM is mapped to the same shutdown path, so process managers (systemd, Docker, Kubernetes) that stop the process with SIGTERM by default still get a clean shutdown.

start_api(middlewares: list[Middleware] | None = None, json_encoder: Type[JSONEncoder] | None = None, jinja_extensions: set[Type['Extension']] | None = None) → None[source]

Initializes and starts the API listener component with the provided middlewares.

Parameters:

middlewares (list[Middleware] | None) -- A list of middleware instances, defaults to an empty list if None.

start_broker(middlewares: list[Middleware] | None = None) → None[source]

Initializes and starts the broker component with the provided middlewares.

Parameters:

middlewares (list[Middleware] | None) -- A list of middleware instances.

start_connector(router_id: UUID | None = None) → None[source]

Initializes and starts the connector component with an optional router ID.

Parameters:

router_id (UUID | None) -- The UUID of the router, defaults to None.

start_grpc(host: str | None = None, port: int | None = None) → None[source]

Initializes and starts the gRPC listener component, EDRI's gRPC API front end (per-event unary rpc plus one shared bidi-streaming rpc, schema generated at runtime from the same @api-decorated events REST/WS/HTML already use). Runs as its own process, sharing the same Broker (started here if it hasn't been already) - can be started instead of, or alongside, start_api().

Parameters:

  • host -- Interface to bind, defaults to EDRI_GRPC_HOST/localhost.

  • port -- Port to bind, defaults to EDRI_GRPC_PORT/50051.

start_scheduler(jobs: list[Job] | Mapping[str, Job] | None = None) → None[source]

Initializes and starts the scheduler component with optional jobs.

Jobs may be given as a list, where the scheduler generates an identifier for each job, or as a mapping, where the keys become the job identifiers usable with the scheduler's Update and Cancel events.

Parameters:

jobs (list[Job] | Mapping[str, Job] | None) -- The jobs to schedule, defaults to an empty list if None.

start_store(path: str | PathLike[str] | None = None, *, failure_policy: PersistenceFailurePolicy = PersistenceFailurePolicy.BUFFER, startup_timeout: float | None = 30.0) → None[source]

Initializes and starts the Store, optionally restoring SQLite KV data.

path=None keeps the Store entirely in memory. Persistent writes are asynchronous; BUFFER retries storage failures without blocking events. Persistent mode confirms child-process restoration before returning and raises RuntimeError on bootstrap failure or timeout. startup_timeout=None disables the parent restoration deadline.

start_watcher() → None[source]