Merge pull request #194 from shunichironomura/organize-modules

Move `WatcherGroup` class to `_watcher` module

capsula/_watcher/__init__.py

@@ -1,4 +1,4 @@
-__all__ = ["TimeWatcher", "UncaughtExceptionWatcher", "WatcherBase"]
-from ._base import WatcherBase
+__all__ = ["TimeWatcher", "UncaughtExceptionWatcher", "WatcherBase", "WatcherGroup"]
+from ._base import WatcherBase, WatcherGroup
 from ._exception import UncaughtExceptionWatcher
 from ._time import TimeWatcher

capsula/_watcher/_base.py

@@ -1,15 +1,58 @@
 from __future__ import annotations
 
+import queue
 from abc import abstractmethod
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from contextlib import AbstractContextManager
+from collections.abc import Hashable
+from typing import TYPE_CHECKING, Any, Dict, Generic, OrderedDict, TypeVar
 
+from capsula._backport import AbstractContextManager
 from capsula._capsule import CapsuleItem
 
+if TYPE_CHECKING:
+    from types import TracebackType
+
 
 class WatcherBase(CapsuleItem):
     @abstractmethod
     def watch(self) -> AbstractContextManager[None]:
         raise NotImplementedError
+
+
+_K = TypeVar("_K", bound=Hashable)
+_V = TypeVar("_V", bound=WatcherBase)
+
+
+class WatcherGroup(Generic[_K, _V], AbstractContextManager[Dict[_K, Any]]):
+    def __init__(self, watchers: OrderedDict[_K, _V]) -> None:
+        self.watchers = watchers
+        self.context_manager_stack: queue.LifoQueue[AbstractContextManager[None]] = queue.LifoQueue()
+
+    def __enter__(self) -> dict[_K, Any]:
+        self.context_manager_stack = queue.LifoQueue()
+        cm_dict = {}
+        for key, watcher in reversed(self.watchers.items()):
+            cm = watcher.watch()
+            self.context_manager_stack.put(cm)
+            cm_dict[key] = cm
+            cm.__enter__()
+        return cm_dict
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
+    ) -> bool:
+        suppress_exception = False
+
+        while not self.context_manager_stack.empty():
+            cm = self.context_manager_stack.get(block=False)
+            suppress = bool(cm.__exit__(exc_type, exc_value, traceback))
+            suppress_exception = suppress_exception or suppress
+
+            # If the current context manager handled the exception, we clear the exception info.
+            if suppress:
+                exc_type, exc_value, traceback = None, None, None
+
+        # Return True if any context manager in the stack handled the exception.
+        return suppress_exception

capsula/encapsulator.py

@@ -3,16 +3,14 @@
 import queue
 import threading
 from collections import OrderedDict
-from collections.abc import Hashable
 from itertools import chain
-from typing import TYPE_CHECKING, Any, Dict, Generic, Tuple, TypeVar, Union
+from typing import TYPE_CHECKING, Any, Tuple, Union
 
 from capsula.utils import ExceptionInfo
 
-from ._backport import AbstractContextManager
 from ._capsule import Capsule
 from ._context import ContextBase
-from ._watcher import WatcherBase
+from ._watcher import WatcherBase, WatcherGroup
 from .exceptions import CapsulaError
 
 if TYPE_CHECKING:
@@ -36,46 +34,6 @@ def encapsulate(self) -> Any:
         return self.obj
 
 
-_K = TypeVar("_K", bound=Hashable)
-_V = TypeVar("_V", bound=WatcherBase)
-
-
-class WatcherGroup(Generic[_K, _V], AbstractContextManager[Dict[_K, Any]]):
-    def __init__(self, watchers: OrderedDict[_K, _V]) -> None:
-        self.watchers = watchers
-        self.context_manager_stack: queue.LifoQueue[AbstractContextManager[None]] = queue.LifoQueue()
-
-    def __enter__(self) -> dict[_K, Any]:
-        self.context_manager_stack = queue.LifoQueue()
-        cm_dict = {}
-        for key, watcher in reversed(self.watchers.items()):
-            cm = watcher.watch()
-            self.context_manager_stack.put(cm)
-            cm_dict[key] = cm
-            cm.__enter__()
-        return cm_dict
-
-    def __exit__(
-        self,
-        exc_type: type[BaseException] | None,
-        exc_value: BaseException | None,
-        traceback: TracebackType | None,
-    ) -> bool:
-        suppress_exception = False
-
-        while not self.context_manager_stack.empty():
-            cm = self.context_manager_stack.get(block=False)
-            suppress = bool(cm.__exit__(exc_type, exc_value, traceback))
-            suppress_exception = suppress_exception or suppress
-
-            # If the current context manager handled the exception, we clear the exception info.
-            if suppress:
-                exc_type, exc_value, traceback = None, None, None
-
-        # Return True if any context manager in the stack handled the exception.
-        return suppress_exception
-
-
 class Encapsulator:
     _thread_local = threading.local()
 

pyproject.toml

@@ -68,7 +68,11 @@ line-length = 120
 preview = true
 explicit-preview-rules = true
 
-select = ["ALL", "RUF022"]
+select = [
+    "ALL",
+    "RUF022", # unsorted-dunder-all (preview)
+]
+
 ignore = [
     "PLR2004", # magic numbers
     "S101",    # use of assert