EDRI

The EDRI class is the main orchestrator. A single instance wires together the Router, Managers, Broker, API, Store, Scheduler, Connector, and Watcher into a running application.

Overview

You create one EDRI() in your entry-point script, add Managers, start whatever services you need, then call run(). The class handles startup ordering, graceful shutdown on SIGINT / SIGTERM, and cleanup of all child processes.

from edri import EDRI

edri = EDRI()
edri.add_component(...)
edri.start_api()
edri.run()

API surface

EDRI()

Constructor — takes no arguments (for now).

edri = EDRI()

Initialises the shared multiprocessing.Queue that the Router, Managers, and all internal components use to pass events.

add_component(manager)

Register a ManagerBase subclass.

edri.add_component(MyManager(edri.router_queue))

The Manager will be started when run() is called. Each component's signature is checked for the restart kwargs that the HealthChecker needs; a warning is logged if any are missing.

start_store(path=None, *, failure_policy=PersistenceFailurePolicy.BUFFER, startup_timeout=30.0)

Start the built-in key-value Store (a specialised Manager). Optional — only call if your application needs it.

Pass path="/var/lib/edri/store.db" for asynchronous SQLite persistence and startup restoration. The default path=None keeps all KV data in memory. See Store persistence for failure handling, TTL, shutdown, and trusted-pickle requirements.

Persistent startup waits up to startup_timeout seconds for restoration; increase it for a large database, or use None for no parent deadline.

edri.start_store()

See Specialized managers for usage.

start_scheduler(jobs=None)

Start the Scheduler with optional initial Job instances, given either as a list or as a mapping whose keys become the job identifiers.

edri.start_scheduler(jobs=[...])
edri.start_scheduler(jobs={"heartbeat": ...})

The Scheduler emits events at configured times/intervals. See Specialized managers.

start_broker(middlewares=None)

Start the Broker component that runs the middleware pipeline. Usually called implicitly by start_api().

edri.start_broker(middlewares=[...])

start_api(middlewares=None, json_encoder=None, jinja_extensions=None)

Start the ASGI Listener (API). This is the primary way to expose HTTP endpoints.

edri.start_api()

Parameters:

  • middlewares — list of Middleware instances applied to every request/response.

  • json_encoder — custom json.JSONEncoder subclass (defaults to EDRI's CustomJSONEncoder).

  • jinja_extensions — set of Jinja2 Extension classes for the HTML handler.

If start_broker has not been called yet, start_api calls it first.

start_grpc(host=None, port=None)

Start the gRPC listener — EDRI's gRPC front end, alongside or instead of start_api()'s ASGI Listener. Runs as its own process, sharing the same Broker (started here if it hasn't been already).

edri.start_grpc()

Parameters:

  • host — interface to bind, defaults to EDRI_GRPC_HOST / localhost.

  • port — port to bind, defaults to EDRI_GRPC_PORT / 50051.

If start_broker has not been called yet, start_grpc calls it first. See gRPC.

start_connector(router_id=None)

Connect to the distributed Switch for cross-machine event routing.

from uuid import UUID

edri.start_connector(router_id=UUID("..."))

See Connector and Switch.

start_watcher()

Enable the development hot-reload watcher. Watches source files and restarts component processes when they change.

edri.start_watcher()

See Watcher (Hot Reload).

run()

Enter the main loop. This method:

  1. Creates the Router.

  2. Starts all registered Managers (calls each component's start()).

  3. Runs the Router's event-dispatch loop.

  4. On KeyboardInterrupt (SIGINT) or SIGTERM, shuts down gracefully: quits the Router, stops every Manager, and terminates all child processes (Store, Scheduler, Broker, Listener, Connector).

edri.run()

Startup order

        sequenceDiagram
    participant Main as Main thread
    participant Router
    participant Store
    participant Scheduler
    participant Broker
    participant Listener
    participant Connector
    participant Mgrs as Managers

    Main->>Main: EDRI()
    Main->>Main: add_component() × N
    Main->>Store: start_store() (optional)
    Main->>Scheduler: start_scheduler() (optional)
    Main->>Broker: start_broker() (optional)
    Main->>Listener: start_api() (optional)
    Main->>Connector: start_connector() (optional)
    Main->>Main: start_watcher() (optional)

    Main->>Router: run()
    Router->>Mgrs: component.start(router_queue)
    Router->>Router: event dispatch loop
    Note over Router: waits for KeyboardInterrupt / SIGTERM
    Router-->>Main: quit signal
    Main->>Mgrs: component.quit()
    Main->>(+)Store: terminate()
    Main->>(+)Scheduler: terminate()
    Main->>(+)Broker: terminate()
    Main->>(+)Listener: terminate()
    Main->>(+)Connector: terminate()

Minimal complete example

from datetime import datetime
from edri import EDRI
from edri.api.dataclass.api_event import api, ApiType
from edri.dataclass.event import Event
from edri.dataclass.response import Response, response
from edri.abstract import ManagerBase, request
from http import HTTPMethod


@response
class GreetingResponse(Response):
    message: str
    timestamp: str


@api(url="/hello/{name}", method=HTTPMethod.GET, exclude={ApiType.HTML, ApiType.WS})
class Greeting(Event):
    name: str
    response: GreetingResponse


class GreetingManager(ManagerBase):
    @request
    def greet(self, event: Greeting) -> None:
        event.response.message = f"Hello, {event.name}!"
        event.response.timestamp = datetime.now().isoformat()


if __name__ == "__main__":
    edri = EDRI()
    edri.add_component(GreetingManager(edri.router_queue))
    edri.start_api()
    edri.run()

Distributed example (with Connector)

from uuid import UUID
from edri import EDRI
from edri.abstract import ManagerBase, request

from my_events import RemoteEvent


class LocalManager(ManagerBase):
    @request
    def handle_remote(self, event: RemoteEvent) -> None:
        print(f"Received from remote node: {event.data}")


if __name__ == "__main__":
    edri = EDRI()
    edri.add_component(LocalManager(edri.router_queue))
    edri.start_connector(router_id=UUID("12345678-1234-5678-1234-567812345678"))
    edri.run()

See also