Coverage for src/kwai/core/events/event_router.py: 58%

1"""Module for defining an event router.""" 

2 

3import inspect 

4 

5from dataclasses import dataclass 

6from typing import Any, Callable, Type 

7 

8from loguru import logger 

9 

10from kwai.core.events.event import Event 

11 

12 

13EventCallbackType = Callable[[dict[str, Any]], Any] 

14 

15 

16@dataclass(frozen=True, slots=True, kw_only=True) 

17class EventRouter: 

18 """A router that defines which method must be called on an event.""" 

19 

20 event: Type[Event] 

21 callback: EventCallbackType 

22 

23 async def execute(self, event_data: dict[str, Any]) -> bool: 

24 """Executes the callback.""" 

25 try: 

26 if inspect.iscoroutinefunction(self.callback): 

27 await self.callback(event_data) 

28 else: 

29 self.callback(event_data) 

30 except Exception as ex: 

31 logger.warning(f"The handler raised an exception: {ex!r}") 

32 return False 

33 return True