chore: No more typing.Optional nor typing.Union (gorakhargosh#1056)

* chore: No more `typing.Optional` nor `typing.Union`

* chore: delete missed ruff rule [skip ci]

pyproject.toml

@@ -60,7 +60,6 @@ ignore = [
     "S",
     "TD",
     "TRY003",
-    "UP",
 ]
 fixable = ["ALL"]
 

src/watchdog/events.py

@@ -96,7 +96,6 @@
 import os.path
 import re
 from dataclasses import dataclass, field
-from typing import Optional
 
 from watchdog.utils.patterns import match_any_paths
 
@@ -450,7 +449,7 @@ def dispatch(self, event: FileSystemEvent) -> None:
 class LoggingEventHandler(FileSystemEventHandler):
     """Logs all the events captured."""
 
-    def __init__(self, logger: Optional[logging.Logger] = None) -> None:
+    def __init__(self, logger: logging.Logger | None = None) -> None:
         super().__init__()
         self.logger = logger or logging.root
 

src/watchdog/observers/api.py

@@ -109,7 +109,7 @@ class EventEmitter(BaseThread):
     :param event_filter:
         Collection of event types to emit, or None for no filtering (default).
     :type event_filter:
-        Optional[Iterable[:class:`watchdog.events.FileSystemEvent`]]
+        Iterable[:class:`watchdog.events.FileSystemEvent`] | None
     """
 
     def __init__(self, event_queue, watch, timeout=DEFAULT_EMITTER_TIMEOUT, event_filter=None):
@@ -290,7 +290,7 @@ def schedule(self, event_handler, path, recursive=False, event_filter=None):
         :param event_filter:
             Collection of event types to emit, or None for no filtering (default).
         :type event_filter:
-            Optional[Iterable[:class:`watchdog.events.FileSystemEvent`]]
+            Iterable[:class:`watchdog.events.FileSystemEvent`] | None
         :return:
             An :class:`ObservedWatch` object instance representing
             a watch.

src/watchdog/observers/fsevents2.py

@@ -25,7 +25,6 @@
 import unicodedata
 import warnings
 from threading import Thread
-from typing import Optional
 
 # pyobjc
 import AppKit
@@ -81,7 +80,7 @@ class FSEventsQueue(Thread):
 
     def __init__(self, path):
         Thread.__init__(self)
-        self._queue: queue.Queue[Optional[list[NativeEvent]]] = queue.Queue()
+        self._queue: queue.Queue[list[NativeEvent] | None] = queue.Queue()
         self._run_loop = None
 
         if isinstance(path, bytes):

src/watchdog/observers/inotify.py

@@ -108,7 +108,7 @@ class InotifyEmitter(EventEmitter):
     :param event_filter:
         Collection of event types to emit, or None for no filtering (default).
     :type event_filter:
-        Optional[Iterable[:class:`watchdog.events.FileSystemEvent`]]
+        Iterable[:class:`watchdog.events.FileSystemEvent`] | None
     """
 
     def __init__(self, event_queue, watch, timeout=DEFAULT_EMITTER_TIMEOUT, event_filter=None):
@@ -250,7 +250,7 @@ class InotifyFullEmitter(InotifyEmitter):
     :param event_filter:
         Collection of event types to emit, or None for no filtering (default).
     :type event_filter:
-        Optional[Iterable[:class:`watchdog.events.FileSystemEvent`]]
+        Iterable[:class:`watchdog.events.FileSystemEvent`] | None
 
     """
 

src/watchdog/observers/inotify_buffer.py

@@ -15,7 +15,6 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING, Union
 
 from watchdog.observers.inotify_c import Inotify, InotifyEvent
 from watchdog.utils import BaseThread
@@ -54,7 +53,7 @@ def close(self):
 
     def _group_events(self, event_list):
         """Group any matching move events"""
-        grouped: list[Union[InotifyEvent, tuple[InotifyEvent, InotifyEvent]]] = []
+        grouped: list[InotifyEvent | tuple[InotifyEvent, InotifyEvent]] = []
         for inotify_event in event_list:
             logger.debug("in-event %s", inotify_event)
 
@@ -65,9 +64,6 @@ 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/observers/kqueue.py

@@ -404,7 +404,7 @@ class KqueueEmitter(EventEmitter):
     :param event_filter:
         Collection of event types to emit, or None for no filtering (default).
     :type event_filter:
-        Optional[Iterable[:class:`watchdog.events.FileSystemEvent`]]
+        Iterable[:class:`watchdog.events.FileSystemEvent`] | None
     :param stat: stat function. See ``os.stat`` for details.
     """
 

src/watchdog/utils/delayed_queue.py

@@ -17,7 +17,7 @@
 import threading
 import time
 from collections import deque
-from typing import Callable, Generic, Optional, TypeVar
+from typing import Callable, Generic, TypeVar
 
 T = TypeVar("T")
 
@@ -45,7 +45,7 @@ def close(self):
         self._not_empty.notify()
         self._not_empty.release()
 
-    def get(self) -> Optional[T]:
+    def get(self) -> T | None:
         """Remove and return an element from the queue, or this queue has been
         closed raise the Closed exception.
         """
@@ -74,7 +74,7 @@ def get(self) -> Optional[T]:
                     self._queue.popleft()
                     return head
 
-    def remove(self, predicate: Callable[[T], bool]) -> Optional[T]:
+    def remove(self, predicate: Callable[[T], bool]) -> T | None:
         """Remove and return the first items for which predicate is True,
         ignoring delay.
         """

src/watchdog/utils/dirsnapshot.py

@@ -55,7 +55,7 @@
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
-    from typing import Any, Callable, Optional
+    from typing import Any, Callable
 
 
 class DirectorySnapshotDiff:
@@ -254,7 +254,7 @@ def __init__(
             path: str,
             recursive: bool = True,
             stat: Callable[[str], os.stat_result] = os.stat,
-            listdir: Callable[[Optional[str]], Iterator[os.DirEntry]] = os.scandir,
+            listdir: Callable[[str | None], Iterator[os.DirEntry]] = os.scandir,
             ignore_device: bool = False,
         ):
             self.path = path
@@ -310,7 +310,7 @@ def __init__(
         path: str,
         recursive: bool = True,
         stat: Callable[[str], os.stat_result] = os.stat,
-        listdir: Callable[[Optional[str]], Iterator[os.DirEntry]] = os.scandir,
+        listdir: Callable[[str | None], Iterator[os.DirEntry]] = os.scandir,
     ):
         self.recursive = recursive
         self.stat = stat
@@ -359,7 +359,7 @@ def paths(self) -> set[str]:
         """Set of file/directory paths in the snapshot."""
         return set(self._stat_info.keys())
 
-    def path(self, uid: tuple[int, int]) -> Optional[str]:
+    def path(self, uid: tuple[int, int]) -> str | None:
         """Returns path for id. None if id is unknown to this snapshot."""
         return self._inode_to_path.get(uid)
 

tests/utils.py

@@ -3,13 +3,13 @@
 import dataclasses
 import os
 from queue import Empty, Queue
-from typing import Optional, Type, Union, Protocol
+from typing import Protocol
 
 from watchdog.events import FileSystemEvent
 from watchdog.observers.api import EventEmitter, ObservedWatch
 from watchdog.utils import platform
 
-Emitter: Type[EventEmitter]
+Emitter: type[EventEmitter]
 
 if platform.is_linux():
     from watchdog.observers.inotify import InotifyEmitter as Emitter
@@ -30,7 +30,7 @@ def __call__(self, *args: str) -> str:
 class StartWatching(Protocol):
     def __call__(
         self,
-        path: Optional[Union[str, bytes]] = ...,
+        path: bytes | str | None = ...,
         use_full_emitter: bool = ...,
         recursive: bool = ...,
     ) -> EventEmitter:
@@ -56,7 +56,7 @@ def joinpath(self, *args: str) -> str:
 
     def start_watching(
         self,
-        path: Optional[Union[str, bytes]] = None,
+        path: bytes | str | None = None,
         use_full_emitter: bool = False,
         recursive: bool = True,
     ) -> EventEmitter: