1""":module: watchdog.utils
2:synopsis: Utility classes and functions.
3:author: [email protected] (Yesudeep Mangalapilly)
4:author: [email protected] (Mickaël Schoentgen)
5
6Classes
7-------
8.. autoclass:: BaseThread
9   :members:
10   :show-inheritance:
11   :inherited-members:
12
13"""
14
15from __future__ import annotations
16
17import sys
18import threading
19from typing import TYPE_CHECKING
20
21if TYPE_CHECKING:
22    from types import ModuleType
23
24    from watchdog.tricks import Trick
25
26
27class UnsupportedLibcError(Exception):
28    pass
29
30
31class WatchdogShutdownError(Exception):
32    """Semantic exception used to signal an external shutdown event."""
33
34
35class BaseThread(threading.Thread):
36    """Convenience class for creating stoppable threads."""
37
38    def __init__(self) -> None:
39        threading.Thread.__init__(self)
40        if hasattr(self, "daemon"):
41            self.daemon = True
42        else:
43            self.setDaemon(True)
44        self._stopped_event = threading.Event()
45
46    @property
47    def stopped_event(self) -> threading.Event:
48        return self._stopped_event
49
50    def should_keep_running(self) -> bool:
51        """Determines whether the thread should continue running."""
52        return not self._stopped_event.is_set()
53
54    def on_thread_stop(self) -> None:
55        """Override this method instead of :meth:`stop()`.
56        :meth:`stop()` calls this method.
57
58        This method is called immediately after the thread is signaled to stop.
59        """
60
61    def stop(self) -> None:
62        """Signals the thread to stop."""
63        self._stopped_event.set()
64        self.on_thread_stop()
65
66    def on_thread_start(self) -> None:
67        """Override this method instead of :meth:`start()`. :meth:`start()`
68        calls this method.
69
70        This method is called right before this thread is started and this
71        object's run() method is invoked.
72        """
73
74    def start(self) -> None:
75        self.on_thread_start()
76        threading.Thread.start(self)
77
78
79def load_module(module_name: str) -> ModuleType:
80    """Imports a module given its name and returns a handle to it."""
81    try:
82        __import__(module_name)
83    except ImportError as e:
84        error = f"No module named {module_name}"
85        raise ImportError(error) from e
86    return sys.modules[module_name]
87
88
89def load_class(dotted_path: str) -> type[Trick]:
90    """Loads and returns a class definition provided a dotted path
91    specification the last part of the dotted path is the class name
92    and there is at least one module name preceding the class name.
93
94    Notes
95    -----
96    You will need to ensure that the module you are trying to load
97    exists in the Python path.
98
99    Examples
100    --------
101    - module.name.ClassName    # Provided module.name is in the Python path.
102    - module.ClassName         # Provided module is in the Python path.
103
104    What won't work:
105    - ClassName
106    - modle.name.ClassName     # Typo in module name.
107    - module.name.ClasNam      # Typo in classname.
108
109    """
110    dotted_path_split = dotted_path.split(".")
111    if len(dotted_path_split) <= 1:
112        error = f"Dotted module path {dotted_path} must contain a module name and a classname"
113        raise ValueError(error)
114    klass_name = dotted_path_split[-1]
115    module_name = ".".join(dotted_path_split[:-1])
116
117    module = load_module(module_name)
118    if hasattr(module, klass_name):
119        return getattr(module, klass_name)
120
121    error = f"Module {module_name} does not have class attribute {klass_name}"
122    raise AttributeError(error)
123