Coverage for src / lexigram / admin / events / adapter.py: 90%

1"""Adapter that delegates admin event dispatch to an ``EventBusProtocol`` bus. 

2 

3``AdminEventBusAdapter`` wraps a container-provided event bus (typically a 

4``lexigram-events`` implementation) for pub/sub event dispatch. When no 

5bus is configured, it falls back to a simple in-process dispatcher. 

6""" 

7 

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING, Any 

11 

12from lexigram.contracts.events import EventBusProtocol 

13 

14if TYPE_CHECKING: 

15 from collections.abc import Callable 

16 

17 

18class _SimpleDispatcher: 

19 """Fallback in-process event dispatcher when no bus is configured.""" 

20 

21 def __init__(self) -> None: 

22 self._handlers: dict[str, list[Callable[..., Any]]] = {} 

23 

24 def subscribe(self, event_type: str, handler: Callable[..., Any]) -> None: 

25 """Register a handler for an event type.""" 

26 self._handlers.setdefault(event_type, []).append(handler) 

27 

28 async def publish(self, event: Any) -> None: 

29 """Dispatch an event to all registered handlers.""" 

30 event_type = type(event).__name__ 

31 for handler in self._handlers.get(event_type, []): 

32 try: 

33 result = handler(event) 

34 if hasattr(result, "__await__"): 

35 await result 

36 except Exception: # noqa: BLE001 

37 pass 

38 

39 

40class AdminEventBusAdapter: 

41 """Bridge between admin's event dispatch and an ``EventBusProtocol`` bus. 

42 

43 Usage:: 

44 

45 bus = AdminEventBusAdapter(event_bus=real_bus) 

46 await bus.publish(AdminStarted(version="1.0")) 

47 """ 

48 

49 def __init__( 

50 self, 

51 event_bus: EventBusProtocol | None = None, 

52 ) -> None: 

53 self._bus: Any = event_bus or _SimpleDispatcher() 

54 

55 async def publish(self, event: Any) -> Any: 

56 """Publish an event to all registered handlers. 

57 

58 Args: 

59 event: The event instance to publish. 

60 

61 Returns: 

62 Dispatch result from the bus if available, else None. 

63 """ 

64 if self._bus is None: 

65 return None 

66 try: 

67 return await self._bus.publish(event) 

68 except Exception: # noqa: BLE001 

69 return None 

70 

71 def subscribe(self, event_type: str, handler: Callable[..., Any]) -> None: 

72 """Register a handler for a given event type. 

73 

74 Args: 

75 event_type: The event type name to subscribe to. 

76 handler: Callable that accepts the event. 

77 """ 

78 if self._bus is not None and hasattr(self._bus, "subscribe"): 

79 self._bus.subscribe(event_type, handler) 

80 

81 async def publish_many(self, events: list[Any]) -> list[Any]: 

82 """Publish multiple events in sequence. 

83 

84 Args: 

85 events: List of event instances. 

86 

87 Returns: 

88 List of dispatch results. 

89 """ 

90 if self._bus is None: 

91 return [] 

92 results: list[Any] = [] 

93 for event in events: 

94 result = await self.publish(event) 

95 results.append(result) 

96 return results 

97 

98 

99__all__ = ["AdminEventBusAdapter"]