Add event-driven dashboard stats with product event tracking

framework/core/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "fastapi>=0.115",
     "pydantic>=2.0",
     "pydantic-settings>=2.0",
+    "pyee>=12.0",
 ]
 
 [build-system]

framework/core/simple_module_core/events.py

@@ -1,14 +1,15 @@
-"""Async in-process event bus for inter-module communication."""
+"""Async in-process event bus backed by pyee for inter-module communication."""
 
 from __future__ import annotations
 
 import asyncio
 import logging
-from collections import defaultdict
 from collections.abc import Callable, Coroutine
 from dataclasses import dataclass
 from typing import Any
 
+from pyee.asyncio import AsyncIOEventEmitter
+
 logger = logging.getLogger(__name__)
 
 EventHandler = Callable[..., Coroutine[Any, Any, None]]
@@ -28,27 +29,43 @@ class ProductCreated(Event):
 
 
 class EventBus:
-    """Simple async event bus.
+    """Async event bus backed by pyee's ``AsyncIOEventEmitter``.
 
     Modules subscribe to event types in ``register_event_handlers``.
     Publishing dispatches to all subscribers concurrently.
+
+    * ``publish``      — awaits all handlers via ``asyncio.gather`` (error-isolated).
+    * ``publish_nowait`` — fire-and-forget via pyee's event-loop scheduling.
     """
 
     def __init__(self) -> None:
-        self._handlers: dict[type[Event], list[EventHandler]] = defaultdict(list)
+        self._emitter = AsyncIOEventEmitter()
+        self._emitter.on("error", self._on_emitter_error)
+
+    @staticmethod
+    def _on_emitter_error(error: Exception) -> None:
+        logger.error("EventBus background handler error: %s", error, exc_info=error)
+
+    @staticmethod
+    def _event_key(event_type: type[Event]) -> str:
+        return f"{event_type.__module__}.{event_type.__qualname__}"
 
     def subscribe(self, event_type: type[Event], handler: EventHandler) -> None:
         """Register a handler for an event type."""
-        self._handlers[event_type].append(handler)
+        self._emitter.on(self._event_key(event_type), handler)
         logger.debug(
             "Subscribed %s to %s",
             getattr(handler, "__qualname__", repr(handler)),
             event_type.__name__,
         )
 
     async def publish(self, event: Event) -> None:
-        """Dispatch event to all registered handlers (awaited)."""
-        handlers = self._handlers.get(type(event), [])
+        """Dispatch event to all registered handlers (awaited).
+
+        All handlers run concurrently via ``asyncio.gather``.
+        Individual handler failures are logged but do not propagate.
+        """
+        handlers = self._emitter.listeners(self._event_key(type(event)))
         if not handlers:
             return
         results = await asyncio.gather(
@@ -66,6 +83,9 @@ async def publish(self, event: Event) -> None:
                 )
 
     def publish_nowait(self, event: Event) -> None:
-        """Fire-and-forget: schedule event dispatch on the current event loop."""
-        loop = asyncio.get_event_loop()
-        loop.create_task(self.publish(event))
+        """Fire-and-forget: schedule event dispatch on the current event loop.
+
+        Uses pyee's ``AsyncIOEventEmitter.emit`` which schedules async
+        handlers as tasks on the running loop.
+        """
+        self._emitter.emit(self._event_key(type(event)), event)

framework/core/tests/test_core.py

@@ -479,6 +479,71 @@ async def handler(e):
         assert len(received) == 1
         assert received[0].order_id == 99
 
+    async def test_subclass_events_do_not_match_parent_subscription(self):
+        """Subscribing to a base Event class should not receive subclass events."""
+        bus = EventBus()
+
+        @dataclass
+        class Parent(Event):
+            pass
+
+        @dataclass
+        class Child(Parent):
+            pass
+
+        calls: list = []
+
+        async def parent_handler(e):
+            calls.append(("parent", e))
+
+        bus.subscribe(Parent, parent_handler)
+        await bus.publish(Child())
+
+        # Child events should not trigger Parent handlers — strict type match.
+        assert calls == []
+
+    async def test_publish_with_no_subscribers_returns_none(self):
+        """publish() should resolve to None when nothing is listening."""
+        bus = EventBus()
+
+        @dataclass
+        class Orphan(Event):
+            pass
+
+        result = await bus.publish(Orphan())
+        assert result is None
+
+    async def test_publish_nowait_with_no_subscribers_is_noop(self):
+        """publish_nowait() on an unheard event should not raise."""
+        bus = EventBus()
+
+        @dataclass
+        class Orphan(Event):
+            pass
+
+        bus.publish_nowait(Orphan())  # must not raise
+
+    async def test_handlers_dispatched_concurrently(self):
+        """All handlers for an event should run concurrently via gather."""
+        import asyncio
+
+        bus = EventBus()
+        order: list[str] = []
+
+        async def slow(e):
+            await asyncio.sleep(0.02)
+            order.append("slow")
+
+        async def fast(e):
+            order.append("fast")
+
+        bus.subscribe(OrderCreated, slow)
+        bus.subscribe(OrderCreated, fast)
+        await bus.publish(OrderCreated(order_id=1))
+
+        # "fast" should complete before "slow" because they run concurrently.
+        assert order == ["fast", "slow"]
+
 
 # ── MenuRegistry Advanced ───────────────────────────────────────────
 

modules/dashboard/dashboard/endpoints/api.py

@@ -0,0 +1,17 @@
+"""REST API endpoints for the Dashboard module."""
+
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from dashboard.handlers import get_product_event_counts
+
+router = APIRouter()
+
+
+@router.get("/stats")
+async def dashboard_stats() -> dict:
+    """Return dashboard statistics including product event counts."""
+    return {
+        "product_events": get_product_event_counts(),
+    }

modules/dashboard/dashboard/handlers.py

@@ -0,0 +1,46 @@
+"""Event handlers for the Dashboard module.
+
+Subscribes to product domain events to maintain real-time stats
+without direct coupling to the Products module's internals.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from products.contracts.events import ProductCreated, ProductDeleted, ProductUpdated
+
+logger = logging.getLogger(__name__)
+
+_product_event_counts: dict[str, int] = {
+    "created": 0,
+    "updated": 0,
+    "deleted": 0,
+}
+
+
+async def on_product_created(event: ProductCreated) -> None:
+    _product_event_counts["created"] += 1
+    logger.info("Dashboard received ProductCreated: %s (id=%d)", event.name, event.product_id)
+
+
+async def on_product_updated(event: ProductUpdated) -> None:
+    _product_event_counts["updated"] += 1
+    logger.info("Dashboard received ProductUpdated: %s (id=%d)", event.name, event.product_id)
+
+
+async def on_product_deleted(event: ProductDeleted) -> None:
+    _product_event_counts["deleted"] += 1
+    logger.info("Dashboard received ProductDeleted: id=%d", event.product_id)
+
+
+def get_product_event_counts() -> dict[str, int]:
+    """Return a snapshot of product event counts."""
+    return dict(_product_event_counts)
+
+
+def reset_product_event_counts() -> None:
+    """Reset counters — useful for testing."""
+    _product_event_counts["created"] = 0
+    _product_event_counts["updated"] = 0
+    _product_event_counts["deleted"] = 0

modules/dashboard/dashboard/module.py

@@ -3,19 +3,27 @@
 from __future__ import annotations
 
 from fastapi import APIRouter
+from products.contracts.events import ProductCreated, ProductDeleted, ProductUpdated
+from simple_module_core.events import EventBus
 from simple_module_core.menu import MenuItem, MenuRegistry, MenuSection
 from simple_module_core.module import ModuleBase, ModuleMeta
 
+from dashboard.handlers import on_product_created, on_product_deleted, on_product_updated
+
 
 class DashboardModule(ModuleBase):
     meta = ModuleMeta(
         name="Dashboard",
+        route_prefix="/api/dashboard",
         view_prefix="",
+        depends_on=["Products"],
     )
 
     def register_routes(self, api_router: APIRouter, view_router: APIRouter) -> None:
+        from dashboard.endpoints.api import router as api
         from dashboard.endpoints.views import router as views
 
+        api_router.include_router(api)
         view_router.include_router(views)
 
     def register_menu_items(self, registry: MenuRegistry) -> None:
@@ -28,3 +36,8 @@ def register_menu_items(self, registry: MenuRegistry) -> None:
                 section=MenuSection.SIDEBAR,
             )
         )
+
+    def register_event_handlers(self, bus: EventBus) -> None:
+        bus.subscribe(ProductCreated, on_product_created)
+        bus.subscribe(ProductUpdated, on_product_updated)
+        bus.subscribe(ProductDeleted, on_product_deleted)

modules/dashboard/pyproject.toml

@@ -10,6 +10,7 @@ dependencies = [
     "simple-module-core",
     "simple-module-db",
     "simple-module-hosting",
+    "products",
 ]
 
 [project.entry-points.simple_module]
@@ -23,3 +24,4 @@ build-backend = "hatchling.build"
 simple-module-core = { workspace = true }
 simple-module-db = { workspace = true }
 simple-module-hosting = { workspace = true }
+products = { workspace = true }