morington

Untitled

Jan 22nd, 2024
1,183
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 2.71 KB | None | 0 0
  1. from typing import Callable, List
  2.  
  3. class Filter:
  4.     def __init__(self, filter_name: str) -> None:
  5.         self.filter_name = filter_name
  6.  
  7.     def match(self, event_type: str, data: str) -> bool:
  8.         raise NotImplementedError("Subclasses must implement the match method")
  9.  
  10. class Router:
  11.     def __init__(self):
  12.         self.handlers = {}
  13.         self.child_routers = []
  14.  
  15.     def callback_query(self, event_filter: Filter) -> Callable:
  16.         def decorator(func: Callable):
  17.             self.handlers[event_filter] = func
  18.             return func
  19.         return decorator
  20.  
  21.     def add_router(self, child_router) -> None:
  22.         if isinstance(child_router, Router):
  23.             self.child_routers.append(child_router)
  24.  
  25. class Dispatcher:
  26.     def __init__(self, routers: List[Router]):
  27.         self.routers = routers
  28.  
  29.     async def process_event(self, event_type: str, data: str):
  30.         for router in self.routers:
  31.             print(router)
  32.             return await self._process_router(router, event_type, data)
  33.  
  34.     async def _process_router(self, router: Router, event_type: str, data: str):
  35.         for event_filter, handler in router.handlers.items():
  36.             if event_filter.match(event_type, data):
  37.                 await handler(event_type, data)
  38.  
  39.         for child_router in router.child_routers:
  40.             await self._process_router(child_router, event_type, data)
  41.  
  42. class EventTypeFilter(Filter):
  43.     def __init__(self, expected_event_type: str) -> None:
  44.         super().__init__("event_type")
  45.         self.expected_event_type = expected_event_type
  46.  
  47.     def match(self, event_type: str, data: str) -> bool:
  48.         return event_type == self.expected_event_type
  49.  
  50. router1 = Router()
  51. router2 = Router()
  52. router3 = Router()
  53.  
  54. @router1.callback_query(EventTypeFilter("task"))
  55. async def create_post(event_type: str, data: str):
  56.     print(f"Handling event in router1: {event_type}, data: {data}")
  57.  
  58. @router2.callback_query(EventTypeFilter("post"))
  59. async def post(event_type: str, data: str):
  60.     print(f"Handling event in router2: {event_type}, data: {data}")
  61.  
  62. @router3.callback_query(EventTypeFilter("uuid"))
  63. async def update(event_type: str, data: str):
  64.     print(f"Handling event in router3: {event_type}, data: {data}")
  65.  
  66. router1.add_router(router2)
  67. router1.add_router(router3)
  68.  
  69. async def main():
  70.     routers = [router1, router2, router3]
  71.     dispatcher = Dispatcher(routers)
  72.     # print(dispatcher.routers)
  73.     # for r in dispatcher.routers:
  74.     #     print(r.handlers)
  75.  
  76.     # Имитация события, которое триггерит хендлер
  77.     await dispatcher.process_event("uuid", "data")
  78.  
  79. if __name__ == "__main__":
  80.     import asyncio
  81.     asyncio.run(main())
  82.  
Advertisement
Add Comment
Please, Sign In to add comment