fix: ruff + mypy

Author: BoboTiG

src/watchdog/observers/inotify.py

@@ -115,8 +115,7 @@
 
 
 class FileSystemEventCtor(Protocol):
-    def __call__(self, src_path: bytes | str, dest_path: bytes | str = "") -> FileSystemEvent:
-        ...
+    def __call__(self, src_path: bytes | str, dest_path: bytes | str = "") -> FileSystemEvent: ...
 
 
 @dataclass
@@ -366,8 +365,12 @@ def _remove_watch_internally(self, wd: WatchDescriptor) -> None:
             self._active_callbacks_by_path[path] = wd2
 
 
-def _select_event_type(dir_event: type[FileSystemEvent], file_event: type[FileSystemEvent], *, is_directory: bool
-                       ) -> FileSystemEventCtor:
+def _select_event_type(
+    dir_event: type[FileSystemEvent],
+    file_event: type[FileSystemEvent],
+    *,
+    is_directory: bool,
+) -> FileSystemEventCtor:
     """selects the correct FileSystemEvent Type based on `is_directory` and returns it."""
     return cast(FileSystemEventCtor, dir_event if is_directory else file_event)
 
@@ -398,7 +401,7 @@ def __init__(
         *,
         timeout: float = DEFAULT_EMITTER_TIMEOUT,
         event_filter: list[type[FileSystemEvent]] | None = None,
-        inotify_fd: InotifyFD | None = None
+        inotify_fd: InotifyFD | None = None,
     ) -> None:
         super().__init__(event_queue, watch, timeout=timeout, event_filter=event_filter)
         self._lock: threading.Lock = threading.Lock()
@@ -413,7 +416,7 @@ def on_thread_start(self) -> None:
             path,
             is_recursive=self.watch.is_recursive,
             event_mask=event_mask,
-            follow_symlink=self.watch.follow_symlink
+            follow_symlink=self.watch.follow_symlink,
         )
 
     def on_thread_stop(self) -> None:
@@ -510,13 +513,16 @@ def get_event_mask_from_filter(self) -> Mask:
             elif cls in {DirCreatedEvent, FileCreatedEvent}:
                 event_mask = Mask(event_mask | InotifyConstants.IN_MOVE | InotifyConstants.IN_CREATE)
             elif cls is DirModifiedEvent:
-                event_mask = Mask(event_mask | (
-                    InotifyConstants.IN_MOVE
-                    | InotifyConstants.IN_ATTRIB
-                    | InotifyConstants.IN_MODIFY
-                    | InotifyConstants.IN_CREATE
-                    | InotifyConstants.IN_CLOSE_WRITE
-                ))
+                event_mask = Mask(
+                    event_mask
+                    | (
+                        InotifyConstants.IN_MOVE
+                        | InotifyConstants.IN_ATTRIB
+                        | InotifyConstants.IN_MODIFY
+                        | InotifyConstants.IN_CREATE
+                        | InotifyConstants.IN_CLOSE_WRITE
+                    )
+                )
             elif cls is FileModifiedEvent:
                 event_mask = Mask(event_mask | InotifyConstants.IN_ATTRIB | InotifyConstants.IN_MODIFY)
             elif cls in {DirDeletedEvent, FileDeletedEvent}:

src/watchdog/observers/inotify_c.py

@@ -33,11 +33,13 @@
 
 inotify_add_watch = cast(
     Callable[[int, bytes, int], WatchDescriptor],
-    ctypes.CFUNCTYPE(c_int, c_int, c_char_p, c_uint32, use_errno=True)(("inotify_add_watch", libc)))
+    ctypes.CFUNCTYPE(c_int, c_int, c_char_p, c_uint32, use_errno=True)(("inotify_add_watch", libc)),
+)
 
 inotify_rm_watch = cast(
     Callable[[int, WatchDescriptor], int],
-    ctypes.CFUNCTYPE(c_int, c_int, c_uint32, use_errno=True)(("inotify_rm_watch", libc)))
+    ctypes.CFUNCTYPE(c_int, c_int, c_uint32, use_errno=True)(("inotify_rm_watch", libc)),
+)
 
 inotify_init = cast(Callable[[], int], ctypes.CFUNCTYPE(c_int, use_errno=True)(("inotify_init", libc)))
 
@@ -124,9 +126,7 @@ class InotifyConstants:
 
 
 def _get_mask_string(mask: int) -> str:
-    return "|".join(
-        name for name, c_val in INOTIFY_ALL_CONSTANTS.items() if mask & c_val
-    )
+    return "|".join(name for name, c_val in INOTIFY_ALL_CONSTANTS.items() if mask & c_val)
 
 
 class InotifyEventStruct(ctypes.Structure):
@@ -173,6 +173,7 @@ def on_watch_deleted(self, wd: WatchDescriptor) -> None:
 @dataclass
 class Watch:
     """Represents an inotify watch"""
+
     wd: WatchDescriptor
     """the inotify watch descriptor"""
     mask: Mask
@@ -194,11 +195,13 @@ def is_used(self) -> bool:
         return bool(self.callbacks)
 
     def short_str(self) -> str:
-        contents = ", ".join([
-            f"wd={self.wd}",
-            f"mask={_get_mask_string(self.mask)}",
-            f"_initial_creation_path={self._initial_creation_path!r}",
-        ])
+        contents = ", ".join(
+            [
+                f"wd={self.wd}",
+                f"mask={_get_mask_string(self.mask)}",
+                f"_initial_creation_path={self._initial_creation_path!r}",
+            ]
+        )
         return f"<{type(self).__name__}: {contents}>"
 
 
@@ -522,7 +525,7 @@ def _parse_event_buffer(event_buffer: bytes) -> Generator[tuple[WatchDescriptor,
         i = 0
         while i + 16 <= len(event_buffer):
             wd, mask, cookie, length = struct.unpack_from("iIII", event_buffer, i)
-            name = event_buffer[i + 16: i + 16 + length].rstrip(b"\0")
+            name = event_buffer[i + 16 : i + 16 + length].rstrip(b"\0")
             i += 16 + length
             yield wd, mask, cookie, name
 
@@ -533,8 +536,7 @@ def _parse_event_buffer(event_buffer: bytes) -> Generator[tuple[WatchDescriptor,
 
 @dataclass(unsafe_hash=True, frozen=True)
 class InotifyEvent:
-    """Inotify event struct wrapper.
-    """
+    """Inotify event struct wrapper."""
 
     wd: WatchDescriptor
     """Watch descriptor"""
@@ -611,10 +613,12 @@ def is_directory(self) -> bool:
         return self.is_delete_self or self.is_move_self or self.mask & InotifyConstants.IN_ISDIR > 0
 
     def __repr__(self) -> str:
-        contents = ", ".join([
-            f"wd={self.wd}",
-            f"mask={_get_mask_string(self.mask)}",
-            f"cookie={self.cookie}",
-            f"name={os.fsdecode(self.name)!r}",
-        ])
+        contents = ", ".join(
+            [
+                f"wd={self.wd}",
+                f"mask={_get_mask_string(self.mask)}",
+                f"cookie={self.cookie}",
+                f"name={os.fsdecode(self.name)!r}",
+            ]
+        )
         return f"<{type(self).__name__}: {contents}>"

src/watchdog/observers/inotify_move_event_grouper.py

@@ -23,6 +23,7 @@
 
 class PathedInotifyEvent(NamedTuple):
     """An InotifyEvent and its full source path"""
+
     ev: InotifyEvent
     path: bytes
 
@@ -79,6 +80,7 @@ def matching_from_event(event: GroupedInotifyEvent) -> bool:
     def get_queued_moved_from_event(self, cookie: int) -> PathedInotifyEvent | None:
         """Finds a queued IN_MOVED_FROM event with the give cookie, but does not
         remove it."""
+
         def matching_from_event(event: GroupedInotifyEvent) -> bool:
             return isinstance(event, PathedInotifyEvent) and event.ev.is_moved_from and event.ev.cookie == cookie
 

src/watchdog/utils/patterns.py

@@ -38,7 +38,7 @@ def _get_sep(path: PurePath) -> str:
 
 def _full_match(path: PurePath, pattern: str) -> bool:
     try:
-        return path.full_match(pattern)
+        return path.full_match(pattern)  # type: ignore[attr-defined]
     except AttributeError:
         # Replicate for python <3.13
         # Please remove this, backwards_compat.py, and python license attributions

tests/test_inotify_watch_group.py

@@ -32,7 +32,7 @@ def create_inotify_watch(path: bytes, *, recursive: bool = False, follow_symlink
         path,
         is_recursive=recursive,
         follow_symlink=follow_symlink,
-        event_mask=WATCHDOG_ALL_EVENTS
+        event_mask=WATCHDOG_ALL_EVENTS,
     )
 
 
@@ -246,6 +246,3 @@ def assert_inotify_root_events(inotify_root: InotifyWatchGroup) -> None:
     inotify_root.deactivate()
     inotify_a.deactivate()
     cleanup()
-
-
-