Merge pull request #816 from flairNLP/fix-duplication-error

Fix `bidict.ValueDuplicationError`

Author: addie9800

src/fundus/scraping/crawler.py

@@ -404,6 +404,7 @@ def callback() -> None:
                         break
         finally:
             session_handler.close_current_session()
+            __EVENTS__.reset()
             if save_to_file is not None:
                 if isinstance(save_to_file, str):
                     save_to_file = Path(save_to_file)

src/fundus/utils/events.py

@@ -1,3 +1,4 @@
+import json
 import threading
 from collections import defaultdict
 from typing import Dict, List, Optional, Union
@@ -84,7 +85,8 @@ def __init__(self, default_events: Optional[List[str]] = None):
             default_events: A list of event names that are automatically available
                 for all threads (e.g., ["stop"]).
         """
-        self._events: Dict[int, ThreadEventDict] = defaultdict(lambda: ThreadEventDict(default_events))
+        self.default_events = default_events
+        self._events: Dict[int, ThreadEventDict] = defaultdict(lambda: ThreadEventDict(self.default_events))
         self._aliases: bidict[str, int] = bidict()
         self._lock = threading.RLock()
 
@@ -163,7 +165,7 @@ def register_event(self, event: str, key: Union[int, str, None] = None):
         with self._lock:
             if isinstance(key, str) and key not in self._aliases:
                 self._alias(key)
-            if (resolved := self._resolve(key)) not in self._events:
+            if event not in self._events[(resolved := self._resolve(key))]:
                 self._events[resolved][event] = threading.Event()
                 logger.debug(f"Registered event {event!r} for {self._pretty_resolve(key)}")
 
@@ -287,5 +289,19 @@ def get(self, event: str, key: Optional[Union[int, str, None]] = None) -> thread
         with self._lock:
             return self._events[self._resolve(key)][event]
 
+    def reset(self):
+        with self._lock:
+            self._events = defaultdict(lambda: ThreadEventDict(self.default_events))
+            self._aliases = bidict()
+
+    def __str__(self):
+        def _entry(thread: int) -> str:
+            alias = self._aliases.inv.get(thread, None)
+            serialized = json.dumps(self._events[thread], indent=2, ensure_ascii=False, default=lambda o: o.set())
+            return f"{alias if alias else 'None'} --> {thread}: \n" + serialized
+
+        events = [_entry(ident) for ident in self._events]
+        return "\n".join(events) if events else "Empty Event Dictionary"
+
 
 __EVENTS__: EventDict = EventDict(default_events=__DEFAULT_EVENTS__)

tests/resources/test_events.py

@@ -0,0 +1,86 @@
+import bidict
+import pytest
+
+from fundus.utils.events import EventDict
+
+
+class TestEvents:
+    def test_default_events(self):
+        events = EventDict(default_events=["success"])
+
+        events.get("success")
+
+        with pytest.raises(KeyError):
+            events.get("failure")
+
+    def test_set_clear(self):
+        events = EventDict(default_events=["success"])
+
+        events.set_event("success")
+
+        assert events.is_event_set("success")
+
+        events.clear_event("success")
+
+        assert events.is_event_set("success") == False
+
+    def test_set_clear_all(self):
+        events = EventDict()
+
+        events.register_event("success", 1)
+        events.register_event("success", 2)
+
+        events.set_for_all("success")
+
+        assert events.is_event_set("success", 1) == events.is_event_set("success", 2) == True
+
+        events.clear_for_all("success")
+
+        assert events.is_event_set("success", 1) == events.is_event_set("success", 2) == False
+
+        events.register_event("failure", 1)
+        events.register_event("failure", 2)
+
+        events.set_for_all()
+
+        assert events.is_event_set("success", 1) == events.is_event_set("success", 2) == True
+        assert events.is_event_set("failure", 1) == events.is_event_set("failure", 2) == True
+
+        events.clear_for_all()
+
+        assert events.is_event_set("success", 1) == events.is_event_set("success", 2) == False
+        assert events.is_event_set("failure", 1) == events.is_event_set("failure", 2) == False
+
+    def test_alias(self):
+        events = EventDict(default_events=["success"])
+
+        events.alias("main-thread")
+
+        events.set_event("success", "main-thread")
+
+        assert events.is_event_set("success", "main-thread") == True
+
+    def test_set_all_with_alias(self):
+        events = EventDict(default_events=["success"])
+
+        events.alias("main-thread")
+
+        events.set_for_all("success")
+
+        assert events.is_event_set("success", "main-thread") == True
+
+        events.clear_for_all("success")
+
+        assert events.is_event_set("success", "main-thread") == False
+
+    def test_duplicate(self):
+        events = EventDict(default_events=["success"])
+
+        events.alias("main-thread", 1)
+
+        with pytest.raises(bidict.ValueDuplicationError):
+            events.alias("new-thread", 1)
+
+        events.alias("main-thread", 2)
+
+        events.alias("new-thread", 1)