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

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 self._global_handlers: list[Callable[..., Any]] = [] 

24 

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

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

27 if event_type is None: 

28 self._global_handlers.append(handler) 

29 return 

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

31 

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

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

34 event_type = type(event).__name__ 

35 for handler in [*self._global_handlers, *self._handlers.get(event_type, [])]: 

36 try: 

37 result = handler(event) 

38 if hasattr(result, "__await__"): 

39 await result 

40 except Exception: # noqa: BLE001, S110 

41 pass 

42 

43 

44class AdminEventBusAdapter: 

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

46 

47 Usage:: 

48 

49 bus = AdminEventBusAdapter(event_bus=real_bus) 

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

51 """ 

52 

53 def __init__( 

54 self, 

55 event_bus: EventBusProtocol | None = None, 

56 ) -> None: 

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

58 

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

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

61 

62 Args: 

63 event: The event instance to publish. 

64 

65 Returns: 

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

67 """ 

68 if self._bus is None: 

69 return None 

70 try: 

71 return await self._bus.publish(event) 

72 except Exception: # noqa: BLE001 

73 return None 

74 

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

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

77 

78 Args: 

79 event_type: The event type name to subscribe to; ``None`` 

80 registers a global handler on the fallback dispatcher. 

81 handler: Callable that accepts the event. 

82 """ 

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

84 self._bus.subscribe(event_type, handler) 

85 

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

87 """Publish multiple events in sequence. 

88 

89 Args: 

90 events: List of event instances. 

91 

92 Returns: 

93 List of dispatch results. 

94 """ 

95 if self._bus is None: 

96 return [] 

97 results: list[Any] = [] 

98 for event in events: 

99 result = await self.publish(event) 

100 results.append(result) 

101 return results 

102 

103 def stream(self) -> Any: 

104 """Return a hot stream of admin events from the configured bus. 

105 

106 Returns: 

107 An ``EventStream[Any]`` from ``lexigram.reactive`` fed by the 

108 bus (or the fallback dispatcher) through a shared Subject. 

109 Live events only; use ``lexigram.events.reactive.from_bus`` 

110 for catchup replay from an event store. 

111 """ 

112 from lexigram.reactive import Subject 

113 

114 subject = Subject[Any]() 

115 

116 async def _handler(event: Any) -> None: 

117 await subject.publish(event) 

118 

119 self.subscribe(None, _handler) 

120 return subject 

121 

122 

123__all__ = ["AdminEventBusAdapter"]