from __future__ import annotations
__all__ = ["EDRI"]
try:
from importlib.metadata import version as _version
__version__ = _version("edri")
except Exception:
__version__ = "0.0.0"
from json import JSONEncoder
from logging import getLogger
from multiprocessing import Queue
from os import PathLike
from typing import TYPE_CHECKING, Type
from uuid import UUID
from edri.dataclass.persistence_failure_policy import PersistenceFailurePolicy
if TYPE_CHECKING:
from collections.abc import Mapping
from edri.api.middleware import Middleware
from edri.api.broker import Broker
from edri.api.handlers.html_handler import HTMLHandler
from edri.api.listener import Listener
from edri.api.grpc.listener import GRPCListener
from edri.abstract import ManagerBase
from edri.config.setting import ENVIRONMENT
from edri.dataclass.event import Event
from edri.router import Router # type: ignore[attr-defined]
from edri.router.connector import Connector # type: ignore[attr-defined]
from edri.utility.manager import Store # type: ignore[attr-defined]
from edri.utility.manager.scheduler import Scheduler, Job
from edri.utility.watcher import Watcher
from jinja2.ext import Extension
[docs]
class EDRI:
"""
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 (:class:`multiprocessing.Queue[Event]`): The queue used for routing events between components.
components (set[:class:`edri.abstract.ManagerBase`]): A set of manager components that are part of the system.
store (:class:`edri.utility.manager.Store`): The store component for data storage.
scheduler (:class:`edri.utility.manager.scheduler.Scheduler`): The scheduler component for scheduling tasks.
api (:class:`edri.api.listener.Listener`): The API listener component.
grpc (:class:`edri.api.grpc.listener.GRPCListener`): The gRPC API listener component.
broker (:class:`edri.api.broker.Broker`): The broker component for handling middleware.
connector (:class:`edri.router.connector.Connector`): The connector component for connecting to external services.
router (:class:`edri.router.Router`): The router component that routes events.
"""
def __init__(self) -> None:
"""
Initializes the EDRI instance, setting up the router queue and initializing component placeholders.
"""
self.logger = getLogger(__name__)
self.router_queue: Queue[Event] = Queue()
self.components: set[ManagerBase] = set()
self.store: Store
self.scheduler: Scheduler
self.api: Listener
self.grpc: GRPCListener
self.broker: Broker
self.connector: Connector
self.router: Router
self.watcher: Watcher
[docs]
def add_component(self, component: ManagerBase) -> None:
"""
Adds a component to the EDRI system.
:param component: The component to be added.
:type component: :class:`edri.abstract.ManagerBase`
"""
from edri.router.health_checker import missing_restart_kwargs
cls = component.__class__
for key in missing_restart_kwargs(cls):
self.logger.warning(
"%s.__init__() does not accept '%s' - deprecated: update its signature "
"to accept it (or **kwargs) so the HealthChecker can restart it with "
"that setting preserved after it crashes.",
cls.__name__, key,
)
self.components.add(component)
[docs]
def start_store(
self, path: str | PathLike[str] | None = None, *,
failure_policy: PersistenceFailurePolicy = PersistenceFailurePolicy.BUFFER,
startup_timeout: float | None = 30.0,
) -> None:
"""
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.
"""
from edri.utility.manager import Store # type: ignore[attr-defined]
store = Store(
self.router_queue, path=path, failure_policy=failure_policy,
startup_timeout=startup_timeout,
)
store.start()
self.store = store
[docs]
def start_scheduler(self, jobs: list[Job] | Mapping[str, Job] | None = None) -> None:
"""
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.
:param jobs: The jobs to schedule, defaults to an empty list if None.
:type jobs: list[Job] | Mapping[str, Job] | None
"""
from edri.utility.manager.scheduler import Scheduler, Job
if not jobs:
jobs = []
self.scheduler = Scheduler(self.router_queue, jobs)
self.scheduler.start()
[docs]
def start_broker(self, middlewares: list[Middleware] | None = None) -> None:
"""
Initializes and starts the broker component with the provided middlewares.
:param middlewares: A list of middleware instances.
:type middlewares: list[Middleware] | None
"""
from edri.api.broker import Broker
from edri.api.middleware import Middleware
if middlewares is None:
middlewares = []
self.broker = Broker(self.router_queue, Queue(), middlewares)
self.broker.start()
[docs]
def start_api(self, middlewares: list[Middleware] | None = None, json_encoder: Type[JSONEncoder] | None = None, jinja_extensions: set[Type["Extension"]] | None = None) -> None:
"""
Initializes and starts the API listener component with the provided middlewares.
:param middlewares: A list of middleware instances, defaults to an empty list if None.
:type middlewares: list[Middleware] | None
"""
from edri.api.listener import Listener
from edri.api.handlers.html_handler import HTMLHandler
from edri.api.middleware import Middleware
if middlewares is None:
middlewares = []
if not hasattr(self, "api_broker"):
self.start_broker(middlewares=middlewares)
else:
self.broker.middlewares_request = [
middleware for middleware in middlewares if middleware.is_request
]
self.broker.middlewares_response = [
middleware for middleware in middlewares if middleware.is_response
]
if json_encoder is None:
from edri.utility.json_encoder import CustomJSONEncoder
json_encoder = CustomJSONEncoder
if jinja_extensions is not None:
HTMLHandler.configure_environment(list(jinja_extensions))
self.api = Listener(self.broker.api_broker_queue, middlewares, json_encoder)
self.api.start()
[docs]
def start_grpc(self, host: str | None = None, port: int | None = None) -> None:
"""
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, :meth:`start_api`.
:param host: Interface to bind, defaults to ``EDRI_GRPC_HOST``/``localhost``.
:param port: Port to bind, defaults to ``EDRI_GRPC_PORT``/``50051``.
"""
from edri.api.grpc.listener import GRPCListener
from edri.config.setting import GRPC_HOST, GRPC_PORT
if not hasattr(self, "broker"):
self.start_broker()
self.grpc = GRPCListener(
self.broker.api_broker_queue,
host=host if host is not None else GRPC_HOST,
port=port if port is not None else GRPC_PORT,
)
self.grpc.start()
[docs]
def start_connector(self, router_id: UUID | None = None) -> None:
"""
Initializes and starts the connector component with an optional router ID.
:param router_id: The UUID of the router, defaults to None.
:type router_id: UUID | None
"""
from edri.router.connector import Connector # type: ignore[attr-defined]
self.connector = Connector(self.router_queue, router_id)
self.connector.start()
[docs]
def start_watcher(self) -> None:
from edri.utility.watcher import Watcher
self.watcher = Watcher(self.router_queue, self.components)
[docs]
def run(self) -> None:
"""
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.
"""
from edri.config.setting import ENVIRONMENT
from edri.router import Router # type: ignore[attr-defined]
from edri.utility.function import install_graceful_sigterm, terminate_processes, COMPONENT_SHUTDOWN_GRACE
self.logger.info("Running EDRI in %s environment", ENVIRONMENT)
install_graceful_sigterm()
self.router = Router(self.router_queue, self.components)
for component in self.components:
component.start(self.router_queue)
try:
self.router.run()
except KeyboardInterrupt:
self.logger.info("Shutting down EDRI")
self.router.quit()
for component in self.components:
component.quit()
if hasattr(self, "watcher") and self.watcher:
self.watcher.quit()
if hasattr(self, "connector") and self.connector:
self.connector.quit()
# Stop every child process EDRI started. They cannot be daemons —
# Managers spawn WorkerProcess children, and Python forbids daemonic
# processes from having children — so the interpreter would otherwise
# block joining them at exit when only the main process receives the
# stop signal (Docker's PID-1 model). terminate() sends SIGTERM, which
# the Managers, Broker and Listener turn into a graceful shutdown; the
# rest stop on it too. (The Watcher is not a process; its quit() above
# already stopped its observer thread.)
#
# COMPONENT_SHUTDOWN_GRACE is deliberately larger than the workers'
# grace so a Manager finishes reaping its own WorkerProcess children
# before we would force-kill it — otherwise a SIGKILL here could land
# mid-reap and orphan a worker.
children = list(self.components)
for attr in ("store", "scheduler", "broker", "api", "grpc", "connector"):
child = getattr(self, attr, None)
if child is not None:
children.append(child)
terminate_processes(children, grace=COMPONENT_SHUTDOWN_GRACE, logger=self.logger)