mypy check_untyped_defs (gorakhargosh#966)

* mypy check_untyped_defs

* maybe fix macos

* flake8

* isort

* flake8

* mypy

* just another approach

* isort

* remove unneeded mypy exception

* cleanup

Author: altendky

mypy.ini

@@ -8,7 +8,7 @@ disallow_subclassing_any = True
 ;disallow_untyped_calls = True
 ;disallow_untyped_defs = True
 disallow_incomplete_defs = True
-;check_untyped_defs = True
+check_untyped_defs = True
 disallow_untyped_decorators = True
 no_implicit_optional = True
 warn_return_any = True

src/watchdog/events.py

@@ -564,17 +564,17 @@ def generate_sub_moved_events(src_dir_path, dest_dir_path):
             renamed_path = (
                 full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
             )
-            event = DirMovedEvent(renamed_path, full_path)
-            event.is_synthetic = True
-            yield event
+            dir_moved_event = DirMovedEvent(renamed_path, full_path)
+            dir_moved_event.is_synthetic = True
+            yield dir_moved_event
         for filename in filenames:
             full_path = os.path.join(root, filename)
             renamed_path = (
                 full_path.replace(dest_dir_path, src_dir_path) if src_dir_path else None
             )
-            event = FileMovedEvent(renamed_path, full_path)
-            event.is_synthetic = True
-            yield event
+            file_moved_event = FileMovedEvent(renamed_path, full_path)
+            file_moved_event.is_synthetic = True
+            yield file_moved_event
 
 
 def generate_sub_created_events(src_dir_path):
@@ -590,10 +590,10 @@ def generate_sub_created_events(src_dir_path):
     """
     for root, directories, filenames in os.walk(src_dir_path):
         for directory in directories:
-            event = DirCreatedEvent(os.path.join(root, directory))
-            event.is_synthetic = True
-            yield event
+            dir_created_event = DirCreatedEvent(os.path.join(root, directory))
+            dir_created_event.is_synthetic = True
+            yield dir_created_event
         for filename in filenames:
-            event = FileCreatedEvent(os.path.join(root, filename))
-            event.is_synthetic = True
-            yield event
+            file_created_event = FileCreatedEvent(os.path.join(root, filename))
+            file_created_event.is_synthetic = True
+            yield file_created_event

src/watchdog/observers/fsevents2.py

@@ -26,6 +26,7 @@
 import unicodedata
 import warnings
 from threading import Thread
+from typing import List, Optional, Type
 
 # pyobjc
 import AppKit  # type: ignore[import]
@@ -65,6 +66,7 @@
     FileDeletedEvent,
     FileModifiedEvent,
     FileMovedEvent,
+    FileSystemEvent,
 )
 from watchdog.observers.api import DEFAULT_EMITTER_TIMEOUT, DEFAULT_OBSERVER_TIMEOUT, BaseObserver, EventEmitter
 
@@ -80,7 +82,7 @@ class FSEventsQueue(Thread):
 
     def __init__(self, path):
         Thread.__init__(self)
-        self._queue = queue.Queue()
+        self._queue: queue.Queue[Optional[List[NativeEvent]]] = queue.Queue()
         self._run_loop = None
 
         if isinstance(path, bytes):
@@ -206,6 +208,7 @@ def queue_events(self, timeout):
         while i < len(events):
             event = events[i]
 
+            cls: Type[FileSystemEvent]
             # For some reason the create and remove flags are sometimes also
             # set for rename and modify type events, so let those take
             # precedence.

src/watchdog/observers/inotify.py

@@ -69,6 +69,7 @@
 import logging
 import os
 import threading
+from typing import Type
 
 from watchdog.events import (
     DirCreatedEvent,
@@ -81,6 +82,7 @@
     FileModifiedEvent,
     FileMovedEvent,
     FileOpenedEvent,
+    FileSystemEvent,
     generate_sub_created_events,
     generate_sub_moved_events,
 )
@@ -128,9 +130,14 @@ def queue_events(self, timeout, full_events=False):
             logger.error("InotifyEmitter.queue_events() called when the thread is inactive")
             return
         with self._lock:
+            if self._inotify is None:
+                logger.error("InotifyEmitter.queue_events() called when the thread is inactive")
+                return
             event = self._inotify.read_event()
             if event is None:
                 return
+
+            cls: Type[FileSystemEvent]
             if isinstance(event, tuple):
                 move_from, move_to = event
                 src_path = self._decode_path(move_from.src_path)

src/watchdog/observers/inotify_buffer.py

@@ -15,8 +15,9 @@
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING, List, Tuple, Union
 
-from watchdog.observers.inotify_c import Inotify
+from watchdog.observers.inotify_c import Inotify, InotifyEvent
 from watchdog.utils import BaseThread
 from watchdog.utils.delayed_queue import DelayedQueue
 
@@ -32,7 +33,7 @@ class InotifyBuffer(BaseThread):
 
     def __init__(self, path, recursive=False):
         super().__init__()
-        self._queue = DelayedQueue(self.delay)
+        self._queue = DelayedQueue[InotifyEvent](self.delay)
         self._inotify = Inotify(path, recursive)
         self.start()
 
@@ -53,7 +54,7 @@ def close(self):
 
     def _group_events(self, event_list):
         """Group any matching move events"""
-        grouped = []
+        grouped: List[Union[InotifyEvent, Tuple[InotifyEvent, InotifyEvent]]] = []
         for inotify_event in event_list:
             logger.debug("in-event %s", inotify_event)
 
@@ -68,6 +69,9 @@ def matching_from_event(event):
                 # Check if move_from is already in the buffer
                 for index, event in enumerate(grouped):
                     if matching_from_event(event):
+                        if TYPE_CHECKING:
+                            # this check is hidden from mypy inside matching_from_event()
+                            assert not isinstance(event, tuple)
                         grouped[index] = (event, inotify_event)
                         break
                 else:

src/watchdog/tricks/__init__.py

@@ -46,6 +46,7 @@
 import os
 import signal
 import subprocess
+import sys
 import threading
 import time
 
@@ -322,7 +323,7 @@ def _restart_process(self):
         self.restart_count += 1
 
 
-if hasattr(os, "getpgid") and hasattr(os, "killpg"):
+if not sys.platform.startswith("win"):
 
     def kill_process(pid, stop_signal):
         os.killpg(os.getpgid(pid), stop_signal)

src/watchdog/utils/delayed_queue.py

@@ -17,17 +17,20 @@
 import threading
 import time
 from collections import deque
+from typing import Callable, Deque, Generic, Optional, Tuple, TypeVar
 
+T = TypeVar("T")
 
-class DelayedQueue:
+
+class DelayedQueue(Generic[T]):
     def __init__(self, delay):
         self.delay_sec = delay
         self._lock = threading.Lock()
         self._not_empty = threading.Condition(self._lock)
-        self._queue = deque()
+        self._queue: Deque[Tuple[T, float, bool]] = deque()
         self._closed = False
 
-    def put(self, element, delay=False):
+    def put(self, element: T, delay: bool = False) -> None:
         """Add element to queue."""
         self._lock.acquire()
         self._queue.append((element, time.time(), delay))
@@ -42,7 +45,7 @@ def close(self):
         self._not_empty.notify()
         self._not_empty.release()
 
-    def get(self):
+    def get(self) -> Optional[T]:
         """Remove and return an element from the queue, or this queue has been
         closed raise the Closed exception.
         """
@@ -71,7 +74,7 @@ def get(self):
                     self._queue.popleft()
                     return head
 
-    def remove(self, predicate):
+    def remove(self, predicate: Callable[[T], bool]) -> Optional[T]:
         """Remove and return the first items for which predicate is True,
         ignoring delay."""
         with self._lock:

src/watchdog/watchmedo.py

@@ -32,7 +32,9 @@
 from argparse import ArgumentParser, RawDescriptionHelpFormatter
 from io import StringIO
 from textwrap import dedent
+from typing import TYPE_CHECKING
 
+from watchdog.observers.api import BaseObserverSubclassCallable
 from watchdog.utils import WatchdogShutdown, load_class
 from watchdog.version import VERSION_STRING
 
@@ -265,11 +267,12 @@ def tricks_from(args):
     """
     Command to execute tricks from a tricks configuration file.
     """
+    Observer: BaseObserverSubclassCallable
     if args.debug_force_polling:
         from watchdog.observers.polling import PollingObserver as Observer
     elif args.debug_force_kqueue:
         from watchdog.observers.kqueue import KqueueObserver as Observer
-    elif args.debug_force_winapi:
+    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and sys.platform.startswith("win")):
         from watchdog.observers.read_directory_changes import WindowsApiObserver as Observer
     elif args.debug_force_inotify:
         from watchdog.observers.inotify import InotifyObserver as Observer
@@ -376,8 +379,8 @@ def tricks_generate_yaml(args):
     else:
         if not os.path.exists(args.append_to_file):
             content = header + content
-        with open(args.append_to_file, "a", encoding="utf-8") as output:
-            output.write(content)
+        with open(args.append_to_file, "a", encoding="utf-8") as file:
+            file.write(content)
 
 
 @command(
@@ -471,11 +474,13 @@ def log(args):
         ignore_patterns=ignore_patterns,
         ignore_directories=args.ignore_directories,
     )
+
+    Observer: BaseObserverSubclassCallable
     if args.debug_force_polling:
         from watchdog.observers.polling import PollingObserver as Observer
     elif args.debug_force_kqueue:
         from watchdog.observers.kqueue import KqueueObserver as Observer
-    elif args.debug_force_winapi:
+    elif (not TYPE_CHECKING and args.debug_force_winapi) or (TYPE_CHECKING and sys.platform.startswith("win")):
         from watchdog.observers.read_directory_changes import WindowsApiObserver as Observer
     elif args.debug_force_inotify:
         from watchdog.observers.inotify import InotifyObserver as Observer
@@ -585,6 +590,7 @@ def shell_command(args):
     if not args.command:
         args.command = None
 
+    Observer: BaseObserverSubclassCallable
     if args.debug_force_polling:
         from watchdog.observers.polling import PollingObserver as Observer
     else:
@@ -704,6 +710,7 @@ def auto_restart(args):
     Command to start a long-running subprocess and restart it on matched events.
     """
 
+    Observer: BaseObserverSubclassCallable
     if args.debug_force_polling:
         from watchdog.observers.polling import PollingObserver as Observer
     else:

tests/test_delayed_queue.py

@@ -23,7 +23,7 @@
 
 @pytest.mark.flaky(max_runs=5, min_passes=1)
 def test_delayed_get():
-    q = DelayedQueue(2)
+    q = DelayedQueue[str](2)
     q.put("", True)
     inserted = time()
     q.get()
@@ -34,7 +34,7 @@ def test_delayed_get():
 
 @pytest.mark.flaky(max_runs=5, min_passes=1)
 def test_nondelayed_get():
-    q = DelayedQueue(2)
+    q = DelayedQueue[str](2)
     q.put("", False)
     inserted = time()
     q.get()

tests/test_inotify_buffer.py

@@ -146,8 +146,9 @@ def delayed(*args, **kwargs):
 class InotifyBufferDelayedRead(InotifyBuffer):
     def run(self, *args, **kwargs):
         # Introduce a delay to trigger the race condition where the file descriptor is
-        # closed prior to a read being triggered.
-        self._inotify.read_events = delay_call(
+        # closed prior to a read being triggered.  Ignoring type concerns since we are
+        # intentionally doing something odd.
+        self._inotify.read_events = delay_call(  # type: ignore[method-assign]
             function=self._inotify.read_events, seconds=1
         )
 

tests/test_snapshot_diff.py

@@ -172,14 +172,15 @@ def test_ignore_device(p):
 
     inode_orig = DirectorySnapshot.inode
 
+    inode_times = 0
+
     def inode(self, path):
         # This function will always return a different device_id,
         # even for the same file.
+        nonlocal inode_times
         result = inode_orig(self, path)
-        inode.times += 1
-        return result[0], result[1] + inode.times
-
-    inode.times = 0
+        inode_times += 1
+        return result[0], result[1] + inode_times
 
     # Set the custom inode function.
     with patch.object(DirectorySnapshot, "inode", new=inode):