Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- from typing import Callable, List
- class Filter:
- def __init__(self, filter_name: str) -> None:
- self.filter_name = filter_name
- def match(self, event_type: str, data: str) -> bool:
- raise NotImplementedError("Subclasses must implement the match method")
- class Router:
- def __init__(self):
- self.handlers = {}
- self.child_routers = []
- def callback_query(self, event_filter: Filter) -> Callable:
- def decorator(func: Callable):
- self.handlers[event_filter] = func
- return func
- return decorator
- def add_router(self, child_router) -> None:
- if isinstance(child_router, Router):
- self.child_routers.append(child_router)
- class Dispatcher:
- def __init__(self, routers: List[Router]):
- self.routers = routers
- async def process_event(self, event_type: str, data: str):
- for router in self.routers:
- print(router)
- return await self._process_router(router, event_type, data)
- async def _process_router(self, router: Router, event_type: str, data: str):
- for event_filter, handler in router.handlers.items():
- if event_filter.match(event_type, data):
- await handler(event_type, data)
- for child_router in router.child_routers:
- await self._process_router(child_router, event_type, data)
- class EventTypeFilter(Filter):
- def __init__(self, expected_event_type: str) -> None:
- super().__init__("event_type")
- self.expected_event_type = expected_event_type
- def match(self, event_type: str, data: str) -> bool:
- return event_type == self.expected_event_type
- router1 = Router()
- router2 = Router()
- router3 = Router()
- @router1.callback_query(EventTypeFilter("task"))
- async def create_post(event_type: str, data: str):
- print(f"Handling event in router1: {event_type}, data: {data}")
- @router2.callback_query(EventTypeFilter("post"))
- async def post(event_type: str, data: str):
- print(f"Handling event in router2: {event_type}, data: {data}")
- @router3.callback_query(EventTypeFilter("uuid"))
- async def update(event_type: str, data: str):
- print(f"Handling event in router3: {event_type}, data: {data}")
- router1.add_router(router2)
- router1.add_router(router3)
- async def main():
- routers = [router1, router2, router3]
- dispatcher = Dispatcher(routers)
- # print(dispatcher.routers)
- # for r in dispatcher.routers:
- # print(r.handlers)
- # Имитация события, которое триггерит хендлер
- await dispatcher.process_event("uuid", "data")
- if __name__ == "__main__":
- import asyncio
- asyncio.run(main())
Advertisement
Add Comment
Please, Sign In to add comment