feat: Improve typing references for events

changelog.rst

@@ -12,6 +12,7 @@ Changelog
 - [core] Enable ``disallow_untyped_calls`` Mypy rule (`#1055 <https://github.com/gorakhargosh/watchdog/pull/1055>`__)
 - [core] Enforced usage of proper keyword-arguments (`#1057 <https://github.com/gorakhargosh/watchdog/pull/1057>`__)
 - [core] Deleted the ``BaseObserverSubclassCallable`` class. Use ``type[BaseObserver]`` directly (`#1055 <https://github.com/gorakhargosh/watchdog/pull/1055>`__)
+- [core] Improve typing references for events (`#1040 <https://github.com/gorakhargosh/watchdog/issues/1040>`__)
 - [inotify] Renamed the ``inotify_event_struct`` class to ``InotifyEventStruct`` (`#1055 <https://github.com/gorakhargosh/watchdog/pull/1055>`__)
 - [inotify] Renamed the ``UnsupportedLibc`` exception to ``UnsupportedLibcError`` (`#1057 <https://github.com/gorakhargosh/watchdog/pull/1057>`__)
 - [watchmedo] Renamed the ``LogLevelException`` exception to ``LogLevelError`` (`#1057 <https://github.com/gorakhargosh/watchdog/pull/1057>`__)

docs/source/examples/patterns.py

@@ -1,7 +1,7 @@
 import sys
 import time
 
-from watchdog.events import PatternMatchingEventHandler
+from watchdog.events import FileSystemEvent, PatternMatchingEventHandler
 from watchdog.observers import Observer
 
 import logging
@@ -10,7 +10,7 @@
 
 
 class MyEventHandler(PatternMatchingEventHandler):
-    def on_any_event(self, event):
+    def on_any_event(self, event: FileSystemEvent):
         logging.debug(event)
 
 

src/watchdog/events.py

@@ -96,6 +96,7 @@
 import os.path
 import re
 from dataclasses import dataclass, field
+from typing import ClassVar
 
 from watchdog.utils.patterns import match_any_paths
 
@@ -205,6 +206,15 @@ class DirMovedEvent(FileSystemMovedEvent):
 class FileSystemEventHandler:
     """Base file system event handler that you can override methods from."""
 
+    dispatch_table: ClassVar = {
+        EVENT_TYPE_CREATED: "on_created",
+        EVENT_TYPE_DELETED: "on_deleted",
+        EVENT_TYPE_MODIFIED: "on_modified",
+        EVENT_TYPE_MOVED: "on_moved",
+        EVENT_TYPE_CLOSED: "on_closed",
+        EVENT_TYPE_OPENED: "on_opened",
+    }
+
     def dispatch(self, event: FileSystemEvent) -> None:
         """Dispatches events to the appropriate methods.
 
@@ -214,14 +224,7 @@ def dispatch(self, event: FileSystemEvent) -> None:
             :class:`FileSystemEvent`
         """
         self.on_any_event(event)
-        {
-            EVENT_TYPE_CREATED: self.on_created,
-            EVENT_TYPE_DELETED: self.on_deleted,
-            EVENT_TYPE_MODIFIED: self.on_modified,
-            EVENT_TYPE_MOVED: self.on_moved,
-            EVENT_TYPE_CLOSED: self.on_closed,
-            EVENT_TYPE_OPENED: self.on_opened,
-        }[event.event_type](event)
+        getattr(self, self.dispatch_table[event.event_type])(event)
 
     def on_any_event(self, event: FileSystemEvent) -> None:
         """Catch-all event handler.
@@ -232,7 +235,7 @@ def on_any_event(self, event: FileSystemEvent) -> None:
             :class:`FileSystemEvent`
         """
 
-    def on_moved(self, event: FileSystemEvent) -> None:
+    def on_moved(self, event: DirMovedEvent | FileMovedEvent) -> None:
         """Called when a file or a directory is moved or renamed.
 
         :param event:
@@ -241,7 +244,7 @@ def on_moved(self, event: FileSystemEvent) -> None:
             :class:`DirMovedEvent` or :class:`FileMovedEvent`
         """
 
-    def on_created(self, event: FileSystemEvent) -> None:
+    def on_created(self, event: DirCreatedEvent | FileCreatedEvent) -> None:
         """Called when a file or directory is created.
 
         :param event:
@@ -250,7 +253,7 @@ def on_created(self, event: FileSystemEvent) -> None:
             :class:`DirCreatedEvent` or :class:`FileCreatedEvent`
         """
 
-    def on_deleted(self, event: FileSystemEvent) -> None:
+    def on_deleted(self, event: DirDeletedEvent | FileDeletedEvent) -> None:
         """Called when a file or directory is deleted.
 
         :param event:
@@ -259,7 +262,7 @@ def on_deleted(self, event: FileSystemEvent) -> None:
             :class:`DirDeletedEvent` or :class:`FileDeletedEvent`
         """
 
-    def on_modified(self, event: FileSystemEvent) -> None:
+    def on_modified(self, event: DirModifiedEvent | FileModifiedEvent) -> None:
         """Called when a file or directory is modified.
 
         :param event:
@@ -268,7 +271,7 @@ def on_modified(self, event: FileSystemEvent) -> None:
             :class:`DirModifiedEvent` or :class:`FileModifiedEvent`
         """
 
-    def on_closed(self, event: FileSystemEvent) -> None:
+    def on_closed(self, event: FileClosedEvent) -> None:
         """Called when a file opened for writing is closed.
 
         :param event:
@@ -277,7 +280,7 @@ def on_closed(self, event: FileSystemEvent) -> None:
             :class:`FileClosedEvent`
         """
 
-    def on_opened(self, event: FileSystemEvent) -> None:
+    def on_opened(self, event: FileOpenedEvent) -> None:
         """Called when a file is opened.
 
         :param event:
@@ -453,36 +456,36 @@ def __init__(self, logger: logging.Logger | None = None) -> None:
         super().__init__()
         self.logger = logger or logging.root
 
-    def on_moved(self, event: FileSystemEvent) -> None:
+    def on_moved(self, event: DirMovedEvent | FileMovedEvent) -> None:
         super().on_moved(event)
 
         what = "directory" if event.is_directory else "file"
         self.logger.info("Moved %s: from %s to %s", what, event.src_path, event.dest_path)
 
-    def on_created(self, event: FileSystemEvent) -> None:
+    def on_created(self, event: DirCreatedEvent | FileCreatedEvent) -> None:
         super().on_created(event)
 
         what = "directory" if event.is_directory else "file"
         self.logger.info("Created %s: %s", what, event.src_path)
 
-    def on_deleted(self, event: FileSystemEvent) -> None:
+    def on_deleted(self, event: DirDeletedEvent | FileDeletedEvent) -> None:
         super().on_deleted(event)
 
         what = "directory" if event.is_directory else "file"
         self.logger.info("Deleted %s: %s", what, event.src_path)
 
-    def on_modified(self, event: FileSystemEvent) -> None:
+    def on_modified(self, event: DirModifiedEvent | FileModifiedEvent) -> None:
         super().on_modified(event)
 
         what = "directory" if event.is_directory else "file"
         self.logger.info("Modified %s: %s", what, event.src_path)
 
-    def on_closed(self, event: FileSystemEvent) -> None:
+    def on_closed(self, event: FileClosedEvent) -> None:
         super().on_closed(event)
 
         self.logger.info("Closed file: %s", event.src_path)
 
-    def on_opened(self, event: FileSystemEvent) -> None:
+    def on_opened(self, event: FileOpenedEvent) -> None:
         super().on_opened(event)
 
         self.logger.info("Opened file: %s", event.src_path)

src/watchdog/tricks/__init__.py

@@ -110,7 +110,7 @@ def __init__(
         self.process = None
         self._process_watchers = set()
 
-    def on_any_event(self, event):
+    def on_any_event(self, event: FileSystemEvent) -> None:
         if event.event_type == EVENT_TYPE_OPENED:
             # FIXME: see issue #949, and find a way to better handle that scenario
             return
@@ -152,8 +152,8 @@ def on_any_event(self, event):
             )
             process_watcher.start()
 
-    def is_process_running(self):
-        return self._process_watchers or (self.process is not None and self.process.poll() is None)
+    def is_process_running(self) -> bool:
+        return bool(self._process_watchers or (self.process is not None and self.process.poll() is None))
 
 
 class AutoRestartTrick(Trick):
@@ -233,7 +233,7 @@ def stop(self):
         if process_watcher is not None:
             process_watcher.join()
 
-    def _start_process(self):
+    def _start_process(self) -> None:
         if self._is_trick_stopping:
             return
 
@@ -243,7 +243,7 @@ def _start_process(self):
             self.process_watcher = ProcessWatcher(self.process, self._restart_process)
             self.process_watcher.start()
 
-    def _stop_process(self):
+    def _stop_process(self) -> None:
         # Ensure the body of the function is not run in parallel in different threads.
         with self._stopping_lock:
             if self._is_process_stopping:
@@ -276,7 +276,7 @@ def _stop_process(self):
             self._is_process_stopping = False
 
     @echo_events
-    def on_any_event(self, event):
+    def on_any_event(self, event: FileSystemEvent) -> None:
         if event.event_type == EVENT_TYPE_OPENED:
             # FIXME: see issue #949, and find a way to better handle that scenario
             return
@@ -286,20 +286,20 @@ def on_any_event(self, event):
         else:
             self._restart_process()
 
-    def _restart_process(self):
+    def _restart_process(self) -> None:
         if self._is_trick_stopping:
             return
         self._stop_process()
         self._start_process()
         self.restart_count += 1
 
 
-if not platform.is_windows():
+if platform.is_windows():
 
-    def kill_process(pid, stop_signal):
-        os.killpg(os.getpgid(pid), stop_signal)
+    def kill_process(pid: int, stop_signal: int) -> None:
+        os.kill(pid, stop_signal)
 
 else:
 
-    def kill_process(pid, stop_signal):
-        os.kill(pid, stop_signal)
+    def kill_process(pid: int, stop_signal: int) -> None:
+        os.killpg(os.getpgid(pid), stop_signal)  # type: ignore[attr-defined]

src/watchdog/utils/__init__.py

@@ -45,7 +45,7 @@ class WatchdogShutdownError(Exception):
 class BaseThread(threading.Thread):
     """Convenience class for creating stoppable threads."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         threading.Thread.__init__(self)
         if hasattr(self, "daemon"):
             self.daemon = True
@@ -73,15 +73,15 @@ def stop(self):
         self._stopped_event.set()
         self.on_thread_stop()
 
-    def on_thread_start(self):
+    def on_thread_start(self) -> None:
         """Override this method instead of :meth:`start()`. :meth:`start()`
         calls this method.
 
         This method is called right before this thread is started and this
         object's run() method is invoked.
         """
 
-    def start(self):
+    def start(self) -> None:
         self.on_thread_start()
         threading.Thread.start(self)
 

src/watchdog/utils/process_watcher.py

@@ -1,27 +1,30 @@
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 
 from watchdog.utils import BaseThread
 
+if TYPE_CHECKING:
+    import subprocess
+    from typing import Callable
+
 logger = logging.getLogger(__name__)
 
 
 class ProcessWatcher(BaseThread):
-    def __init__(self, popen_obj, process_termination_callback):
+    def __init__(self, popen_obj: subprocess.Popen, process_termination_callback: Callable | None) -> None:
         super().__init__()
         self.popen_obj = popen_obj
         self.process_termination_callback = process_termination_callback
 
-    def run(self):
-        while True:
-            if self.popen_obj.poll() is not None:
-                break
+    def run(self) -> None:
+        while self.popen_obj.poll() is None:
             if self.stopped_event.wait(timeout=0.1):
                 return
 
         try:
-            if not self.stopped_event.is_set():
+            if not self.stopped_event.is_set() and self.process_termination_callback:
                 self.process_termination_callback()
         except Exception:
             logger.exception("Error calling process termination callback")