From e92d67bfe8dbe065250a8d7fe6311e4149d0c488 Mon Sep 17 00:00:00 2001 From: Shreya Malviya Date: Tue, 9 Aug 2022 11:46:19 +0530 Subject: [PATCH] Common: Fix type hints in IEventQueue and PyPubSubEventQueue --- monkey/common/event_queue/i_event_queue.py | 8 +++++--- monkey/common/event_queue/pypubsub_event_queue.py | 8 +++++--- 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/monkey/common/event_queue/i_event_queue.py b/monkey/common/event_queue/i_event_queue.py index 960b5b385..06a99b163 100644 --- a/monkey/common/event_queue/i_event_queue.py +++ b/monkey/common/event_queue/i_event_queue.py @@ -10,7 +10,7 @@ class IEventQueue(ABC): """ @abstractstaticmethod - def subscribe_all(subscriber: Callable[..., Any]): + def subscribe_all(subscriber: Callable[[AbstractEvent], None]): """ Subscribes a subscriber to all events @@ -20,7 +20,9 @@ class IEventQueue(ABC): pass @abstractstaticmethod - def subscribe_types(types: Sequence[AbstractEvent], subscriber: Callable[..., Any]): + def subscribe_types( + types: Sequence[AbstractEvent], subscriber: Callable[[AbstractEvent], None] + ): """ Subscribes a subscriber to all specifed event types @@ -31,7 +33,7 @@ class IEventQueue(ABC): pass @abstractstaticmethod - def subscribe_tags(tags: Sequence[str], subscriber: Callable[..., Any]): + def subscribe_tags(tags: Sequence[str], subscriber: Callable[[AbstractEvent], None]): """ Subscribes a subscriber to all specified event tags diff --git a/monkey/common/event_queue/pypubsub_event_queue.py b/monkey/common/event_queue/pypubsub_event_queue.py index 97b4e71dc..e2e6e8765 100644 --- a/monkey/common/event_queue/pypubsub_event_queue.py +++ b/monkey/common/event_queue/pypubsub_event_queue.py @@ -9,18 +9,20 @@ from .i_event_queue import IEventQueue class PyPubSubEventQueue(IEventQueue): @staticmethod - def subscribe_all(subscriber: Callable[..., Any]): + def subscribe_all(subscriber: Callable[[AbstractEvent], None]): pub.subscribe(listener=subscriber, topicName=pub.ALL_TOPICS) @staticmethod - def subscribe_types(types: Sequence[AbstractEvent], subscriber: Callable[..., Any]): + def subscribe_types( + types: Sequence[AbstractEvent], subscriber: Callable[[AbstractEvent], None] + ): for event_type in types: # pypubsub.pub.subscribe needs a string as the topic/event name event_type_name = event_type.__name__ pub.subscribe(listener=subscriber, topicName=event_type_name) @staticmethod - def subscribe_tags(tags: Sequence[str], subscriber: Callable[..., Any]): + def subscribe_tags(tags: Sequence[str], subscriber: Callable[[AbstractEvent], None]): for tag in tags: pub.subscribe(listener=subscriber, topicName=tag)