Periodic refresh into the Key-Value Store

A typical use of the Scheduler + Store pair: every N seconds, the Scheduler fires a RefreshCatalog event; a manager handles it, computes a fresh value, and writes it into the Store via Set. Other managers read the cached value with Get instead of recomputing.

This pattern keeps expensive lookups out of the request path.

---

Events

# events/catalog.py
from edri.dataclass.event import Event, event
from edri.dataclass.response import Response, response


@response
class CatalogSnapshotResponse(Response):
    items: list[str]


@event
class RefreshCatalog(Event):
    """Internal event — emitted by the Scheduler."""


@event
class GetCatalog(Event):
    """Request the latest cached snapshot."""
    response: CatalogSnapshotResponse

RefreshCatalog has no response; it's a fire-and-forget signal.

---

Manager

The manager has two jobs:

1. Producer — handle RefreshCatalog ticks, recompute, write into the Store.

2. Consumer — handle GetCatalog requests by asking the Store and completing the original event when the Store replies.

EDRI threads the original event through Get.data, so the response handler can fish it back out.

# managers/catalog.py
from edri.abstract import ManagerBase, request, response
from edri.events.edri.store import Get, Set

from events.catalog import GetCatalog, RefreshCatalog


CACHE_KEY = "catalog_snapshot"


class CatalogManager(ManagerBase):

    @request
    def refresh(self, event: RefreshCatalog) -> None:
        items = self._load_items_from_source()
        self.router_queue.put(Set(name=CACHE_KEY, value=items))
        self.logger.debug("Catalog refreshed: %d items", len(items))

    @request
    def serve(self, event: GetCatalog) -> None:
        # Ask the Store for the cached value. The original GetCatalog rides
        # along on Get.data so we can match it up when the reply arrives.
        self.router_queue.put(Get(name=CACHE_KEY, data=event))

    @response
    def on_store_reply(self, event: Get) -> None:
        original: GetCatalog | None = event.data  # type: ignore[assignment]
        if original is None:
            return  # Not for us — some other consumer triggered this Get.
        original.response.items = event.response.value or []
        self.router_queue.put(original)

    def _load_items_from_source(self) -> list[str]:
        # Replace with the real expensive lookup.
        return ["alpha", "beta", "gamma"]

Notes:

• @response declares a handler for replies (here, the Get reply from the Store).

• The Store ignores Get.data — it only fills response.value — but data survives the round trip because it's a normal dataclass field.

• Re-emitting original returns the completed GetCatalog to the router for delivery back to the API client (or whichever caller produced it).

---

Entry point

# main.py
from datetime import datetime, timedelta

from edri import EDRI
from edri.utility.manager.scheduler import Job

from events.catalog import RefreshCatalog
from managers.catalog import CatalogManager

if __name__ == "__main__":
    edri = EDRI()

    # Specialized managers.
    edri.start_store()
    edri.start_scheduler(jobs=[
        Job(
            event=RefreshCatalog(),
            when=datetime.now(),                 # fire immediately on start
            repeat=timedelta(seconds=30),        # then every 30 s
        ),
    ])

    # Application manager.
    edri.add_component(CatalogManager(edri.router_queue))

    edri.run()

What happens at runtime:

1. start_store() brings up the Key-Value Store.

2. start_scheduler(jobs=[...]) registers the recurring RefreshCatalog job.

3. On the first tick (and every 30 s after), the Scheduler emits RefreshCatalog.

4. CatalogManager.refresh computes a new snapshot and writes it to the Store.

5. A consumer emits GetCatalog; CatalogManager.serve asks the Store; on_store_reply returns the completed event.

---

Scheduling jobs at runtime

You don't have to declare jobs at startup. From inside a manager, send a Set event to the scheduler:

from edri.events.edri.scheduler import Set as SchedulerSet

self.router_queue.put(
    SchedulerSet(
        event=RefreshCatalog(),
        when=datetime.now() + timedelta(minutes=5),
        repeat=timedelta(hours=1),
        identifier="catalog-refresh",
    )
)

Cancel, Update, and SetUpdate are documented in Scheduler control events.