diff --git a/.cirrus.yml b/.cirrus.yml index b305a1ad..186765ff 100644 --- a/.cirrus.yml +++ b/.cirrus.yml @@ -6,16 +6,16 @@ task: image_family: freebsd-12-2 install_script: - - pkg install -y python38 py38-sqlite3 + - pkg install -y python39 py39-sqlite3 # Print the Python version, only to be sure we are running the version we want - - python3.8 -c 'import platform; print("Python", platform.python_version())' + - python3.9 -c 'import platform; print("Python", platform.python_version())' # Check SQLite3 is installed - - python3.8 -c 'import sqlite3; print("SQLite3", sqlite3.version)' + - python3.9 -c 'import sqlite3; print("SQLite3", sqlite3.version)' setup_script: - - python3.8 -m ensurepip - - python3.8 -m pip install -U pip - - python3.8 -m pip install -r requirements-tests.txt + - python3.9 -m ensurepip + - python3.9 -m pip install -U pip + - python3.9 -m pip install -r requirements-tests.txt lint_script: - - python3.8 -m ruff src + - python3.9 -m ruff src tests_script: - - python3.8 -bb -m pytest tests + - python3.9 -bb -m pytest tests diff --git a/.github/workflows/build-and-publish.yml b/.github/workflows/build-and-publish.yml index de4371f4..d05f8790 100644 --- a/.github/workflows/build-and-publish.yml +++ b/.github/workflows/build-and-publish.yml @@ -53,7 +53,6 @@ jobs: - name: Build wheels run: python -m cibuildwheel env: - CIBW_SKIP: "cp36-*" # skip 3.6 wheels CIBW_ARCHS_MACOS: "x86_64 universal2 arm64" - name: Artifacts list run: ls -l wheelhouse diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 9dd02e0f..6afed287 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -39,13 +39,11 @@ jobs: emoji: 🪟 runs-on: [windows-latest] python: - - "3.8" - "3.9" - "3.10" - "3.11" - "3.12" - "3.13-dev" - - "pypy-3.8" - "pypy-3.9" include: - tox: @@ -67,15 +65,9 @@ jobs: emoji: 🐧 runs-on: [ubuntu-latest] exclude: - - os: - matrix: macos - python: "pypy-3.8" - os: matrix: macos python: "pypy-3.9" - - os: - matrix: windows - python: "pypy-3.8" - os: matrix: windows python: "pypy-3.9" diff --git a/README.rst b/README.rst index 2d6e0dd4..41656cdf 100755 --- a/README.rst +++ b/README.rst @@ -6,7 +6,7 @@ Watchdog Python API and shell utilities to monitor file system events. -Works on 3.8+. +Works on 3.9+. Example API Usage ----------------- @@ -211,7 +211,7 @@ appropriate observer like in the example above, do:: Dependencies ------------ -1. Python 3.8 or above. +1. Python 3.9 or above. 2. XCode_ (only on macOS when installing from sources) 3. PyYAML_ (only for ``watchmedo``) diff --git a/changelog.rst b/changelog.rst index 467ae4d5..10017dc2 100644 --- a/changelog.rst +++ b/changelog.rst @@ -8,7 +8,11 @@ Changelog 2024-xx-xx • `full history `__ -- +- Drop support for Python 3.8 (`#1055 `__) +- [core] Enable ``disallow_untyped_calls`` Mypy rule (`#1055 `__) +- [core] Deleted the ``BaseObserverSubclassCallable`` class. Use ``type[BaseObserver]`` directly (`#1055 `__) +- [inotify] Renamed the ``inotify_event_struct`` class to ``InotifyEventStruct`` (`#1055 `__) +- [windows] Renamed the ``FILE_NOTIFY_INFORMATION`` class to ``FileNotifyInformation`` (`#1055 `__) - Thanks to our beloved contributors: @BoboTiG 4.0.2 diff --git a/docs/source/index.rst b/docs/source/index.rst index 8365309a..1df5b69d 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -11,7 +11,7 @@ Watchdog Python API library and shell utilities to monitor file system events. -Works on 3.8+. +Works on 3.9+. Directory monitoring made easy with ----------------------------------- diff --git a/docs/source/installation.rst b/docs/source/installation.rst index e1e331d6..a445e553 100644 --- a/docs/source/installation.rst +++ b/docs/source/installation.rst @@ -4,7 +4,7 @@ Installation ============ -|project_name| requires 3.8+ to work. See a list of :ref:`installation-dependencies`. +|project_name| requires 3.9+ to work. See a list of :ref:`installation-dependencies`. Installing from PyPI using pip ------------------------------ diff --git a/pyproject.toml b/pyproject.toml index 9be9550e..63901d21 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -11,7 +11,7 @@ follow_imports = "skip" # Ensure full coverage #disallow_untyped_defs = true [TODO] disallow_incomplete_defs = true -#disallow_untyped_calls = true [TODO] +disallow_untyped_calls = true # Restrict dynamic typing (a little) # e.g. `x: List[Any]` or x: List` @@ -31,7 +31,7 @@ addopts = """ [tool.ruff] line-length = 120 indent-width = 4 -target-version = "py38" +target-version = "py39" [tool.ruff.lint] extend-select = ["ALL"] @@ -53,14 +53,14 @@ ignore = [ "FBT", "FIX", "ISC001", - "N", # Requires a major version number bump + "N818", "PERF203", # TODO "PL", "PTH", # TODO? "S", "TD", "TRY003", - "UP", # TODO when minimum python version will be 3.10 + "UP", ] fixable = ["ALL"] diff --git a/setup.py b/setup.py index d1159e81..923cd532 100644 --- a/setup.py +++ b/setup.py @@ -129,7 +129,6 @@ "Programming Language :: Python", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3 :: Only", - "Programming Language :: Python :: 3.8", "Programming Language :: Python :: 3.9", "Programming Language :: Python :: 3.10", "Programming Language :: Python :: 3.11", @@ -155,6 +154,6 @@ "watchmedo = watchdog.watchmedo:main [watchmedo]", ] }, - python_requires=">=3.8", + python_requires=">=3.9", zip_safe=False, ) diff --git a/src/watchdog/observers/__init__.py b/src/watchdog/observers/__init__.py index 293a6f6b..f1959b2f 100644 --- a/src/watchdog/observers/__init__.py +++ b/src/watchdog/observers/__init__.py @@ -58,10 +58,10 @@ from watchdog.utils import UnsupportedLibc, platform if TYPE_CHECKING: - from watchdog.observers.api import BaseObserverSubclassCallable + from watchdog.observers.api import BaseObserver -def _get_observer_cls() -> BaseObserverSubclassCallable: +def _get_observer_cls() -> type[BaseObserver]: if platform.is_linux(): with contextlib.suppress(UnsupportedLibc): from watchdog.observers.inotify import InotifyObserver diff --git a/src/watchdog/observers/api.py b/src/watchdog/observers/api.py index 4dc7d3c6..df4a683a 100644 --- a/src/watchdog/observers/api.py +++ b/src/watchdog/observers/api.py @@ -20,7 +20,7 @@ import threading from pathlib import Path -from watchdog.utils import BaseThread, Protocol +from watchdog.utils import BaseThread from watchdog.utils.bricks import SkipRepeatsQueue DEFAULT_EMITTER_TIMEOUT = 1 # in seconds. @@ -385,7 +385,3 @@ def dispatch_events(self, event_queue): if handler in self._handlers.get(watch, []): handler.dispatch(event) event_queue.task_done() - - -class BaseObserverSubclassCallable(Protocol): - def __call__(self, timeout: float = ...) -> BaseObserver: ... diff --git a/src/watchdog/observers/fsevents2.py b/src/watchdog/observers/fsevents2.py index 71941742..72e05bfd 100644 --- a/src/watchdog/observers/fsevents2.py +++ b/src/watchdog/observers/fsevents2.py @@ -25,7 +25,7 @@ import unicodedata import warnings from threading import Thread -from typing import List, Optional, Type +from typing import Optional # pyobjc import AppKit @@ -81,7 +81,7 @@ class FSEventsQueue(Thread): def __init__(self, path): Thread.__init__(self) - self._queue: queue.Queue[Optional[List[NativeEvent]]] = queue.Queue() + self._queue: queue.Queue[Optional[list[NativeEvent]]] = queue.Queue() self._run_loop = None if isinstance(path, bytes): @@ -123,9 +123,9 @@ def stop(self): if self._run_loop is not None: CFRunLoopStop(self._run_loop) - def _callback(self, streamRef, clientCallBackInfo, numEvents, eventPaths, eventFlags, eventIDs): - events = [NativeEvent(path, flags, _id) for path, flags, _id in zip(eventPaths, eventFlags, eventIDs)] - logger.debug("FSEvents callback. Got %d events:", numEvents) + def _callback(self, stream_ref, client_callback_info, num_events, event_paths, event_flags, event_ids): + events = [NativeEvent(path, flags, _id) for path, flags, _id in zip(event_paths, event_flags, event_ids)] + logger.debug("FSEvents callback. Got %d events:", num_events) for e in events: logger.debug(e) self._queue.put(events) @@ -195,7 +195,7 @@ def queue_events(self, timeout): while i < len(events): event = events[i] - cls: Type[FileSystemEvent] + 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. diff --git a/src/watchdog/observers/inotify.py b/src/watchdog/observers/inotify.py index f45e339c..a69f9236 100644 --- a/src/watchdog/observers/inotify.py +++ b/src/watchdog/observers/inotify.py @@ -68,7 +68,6 @@ import logging import os import threading -from typing import Type from watchdog.events import ( DirCreatedEvent, @@ -141,7 +140,7 @@ def queue_events(self, timeout, full_events=False): if event is None: return - cls: Type[FileSystemEvent] + cls: type[FileSystemEvent] if isinstance(event, tuple): move_from, move_to = event src_path = self._decode_path(move_from.src_path) diff --git a/src/watchdog/observers/inotify_buffer.py b/src/watchdog/observers/inotify_buffer.py index dbb05aa2..9dc91179 100644 --- a/src/watchdog/observers/inotify_buffer.py +++ b/src/watchdog/observers/inotify_buffer.py @@ -15,7 +15,7 @@ from __future__ import annotations import logging -from typing import TYPE_CHECKING, List, Tuple, Union +from typing import TYPE_CHECKING, Union from watchdog.observers.inotify_c import Inotify, InotifyEvent from watchdog.utils import BaseThread @@ -54,7 +54,7 @@ def close(self): def _group_events(self, event_list): """Group any matching move events""" - grouped: List[Union[InotifyEvent, Tuple[InotifyEvent, InotifyEvent]]] = [] + grouped: list[Union[InotifyEvent, tuple[InotifyEvent, InotifyEvent]]] = [] for inotify_event in event_list: logger.debug("in-event %s", inotify_event) diff --git a/src/watchdog/observers/inotify_c.py b/src/watchdog/observers/inotify_c.py index 0935aaa5..63b2a890 100644 --- a/src/watchdog/observers/inotify_c.py +++ b/src/watchdog/observers/inotify_c.py @@ -113,7 +113,7 @@ class InotifyConstants: ) -class inotify_event_struct(ctypes.Structure): +class InotifyEventStruct(ctypes.Structure): """Structure representation of the inotify_event structure (used in buffer size calculations):: @@ -135,7 +135,7 @@ class inotify_event_struct(ctypes.Structure): ) -EVENT_SIZE = ctypes.sizeof(inotify_event_struct) +EVENT_SIZE = ctypes.sizeof(InotifyEventStruct) DEFAULT_NUM_EVENTS = 2048 DEFAULT_EVENT_BUFFER_SIZE = DEFAULT_NUM_EVENTS * (EVENT_SIZE + 16) diff --git a/src/watchdog/observers/winapi.py b/src/watchdog/observers/winapi.py index b8576523..3e0514cc 100644 --- a/src/watchdog/observers/winapi.py +++ b/src/watchdog/observers/winapi.py @@ -230,7 +230,7 @@ def _errcheck_dword(value, func, args): ) -class FILE_NOTIFY_INFORMATION(ctypes.Structure): +class FileNotifyInformation(ctypes.Structure): _fields_ = ( ("NextEntryOffset", ctypes.wintypes.DWORD), ("Action", ctypes.wintypes.DWORD), @@ -240,7 +240,7 @@ class FILE_NOTIFY_INFORMATION(ctypes.Structure): ) -LPFNI = ctypes.POINTER(FILE_NOTIFY_INFORMATION) +LPFNI = ctypes.POINTER(FileNotifyInformation) # We don't need to recalculate these flags every time a call is made to @@ -281,19 +281,19 @@ class FILE_NOTIFY_INFORMATION(ctypes.Structure): PATH_BUFFER_SIZE = 2048 -def _parse_event_buffer(readBuffer, nBytes): +def _parse_event_buffer(read_buffer, n_bytes): results = [] - while nBytes > 0: - fni = ctypes.cast(readBuffer, LPFNI)[0] - ptr = ctypes.addressof(fni) + FILE_NOTIFY_INFORMATION.FileName.offset + while n_bytes > 0: + fni = ctypes.cast(read_buffer, LPFNI)[0] + ptr = ctypes.addressof(fni) + FileNotifyInformation.FileName.offset # filename = ctypes.wstring_at(ptr, fni.FileNameLength) filename = ctypes.string_at(ptr, fni.FileNameLength) results.append((fni.Action, filename.decode("utf-16"))) - numToSkip = fni.NextEntryOffset - if numToSkip <= 0: + num_to_skip = fni.NextEntryOffset + if num_to_skip <= 0: break - readBuffer = readBuffer[numToSkip:] - nBytes -= numToSkip # numToSkip is long. nBytes should be long too. + read_buffer = read_buffer[num_to_skip:] + n_bytes -= num_to_skip # numToSkip is long. nBytes should be long too. return results @@ -309,7 +309,7 @@ def _is_observed_path_deleted(handle, path): def _generate_observed_path_deleted_event(): # Create synthetic event for notify that observed directory is deleted path = ctypes.create_unicode_buffer(".") - event = FILE_NOTIFY_INFORMATION(0, FILE_ACTION_DELETED_SELF, len(path), path.value.encode("utf-8")) + event = FileNotifyInformation(0, FILE_ACTION_DELETED_SELF, len(path), path.value.encode("utf-8")) event_size = ctypes.sizeof(event) buff = ctypes.create_string_buffer(PATH_BUFFER_SIZE) ctypes.memmove(buff, ctypes.addressof(event), event_size) diff --git a/src/watchdog/utils/__init__.py b/src/watchdog/utils/__init__.py index ece812ff..be62f346 100644 --- a/src/watchdog/utils/__init__.py +++ b/src/watchdog/utils/__init__.py @@ -32,7 +32,6 @@ import sys import threading -from typing import TYPE_CHECKING class UnsupportedLibc(Exception): @@ -130,13 +129,3 @@ def load_class(dotted_path): # return klass(*args, **kwargs) raise AttributeError(f"Module {module_name} does not have class attribute {klass_name}") - - -if TYPE_CHECKING or sys.version_info >= (3, 8): - # using `as` to explicitly re-export this since this is a compatibility layer - from typing import Protocol as Protocol -else: - # Provide a dummy Protocol class when not available from stdlib. Should be used - # only for hinting. This could be had from typing_protocol, but not worth adding - # the _first_ dependency just for this. - class Protocol: ... diff --git a/src/watchdog/utils/delayed_queue.py b/src/watchdog/utils/delayed_queue.py index e6f11836..fbda2a8a 100644 --- a/src/watchdog/utils/delayed_queue.py +++ b/src/watchdog/utils/delayed_queue.py @@ -17,7 +17,7 @@ import threading import time from collections import deque -from typing import Callable, Deque, Generic, Optional, Tuple, TypeVar +from typing import Callable, Generic, Optional, TypeVar T = TypeVar("T") @@ -27,7 +27,7 @@ def __init__(self, delay): self.delay_sec = delay self._lock = threading.Lock() self._not_empty = threading.Condition(self._lock) - self._queue: Deque[Tuple[T, float, bool]] = deque() + self._queue: deque[tuple[T, float, bool]] = deque() self._closed = False def put(self, element: T, delay: bool = False) -> None: diff --git a/src/watchdog/utils/dirsnapshot.py b/src/watchdog/utils/dirsnapshot.py index 77f03187..0e722464 100644 --- a/src/watchdog/utils/dirsnapshot.py +++ b/src/watchdog/utils/dirsnapshot.py @@ -51,7 +51,11 @@ import errno import os from stat import S_ISDIR -from typing import Any, Callable, Iterator, List, Optional, Tuple +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from collections.abc import Iterator + from typing import Any, Callable, Optional class DirectorySnapshotDiff: @@ -90,12 +94,12 @@ def __init__( if ignore_device: - def get_inode(directory: DirectorySnapshot, full_path: str) -> int | Tuple[int, int]: + def get_inode(directory: DirectorySnapshot, full_path: str) -> int | tuple[int, int]: return directory.inode(full_path)[0] else: - def get_inode(directory: DirectorySnapshot, full_path: str) -> int | Tuple[int, int]: + def get_inode(directory: DirectorySnapshot, full_path: str) -> int | tuple[int, int]: return directory.inode(full_path) # check that all unchanged paths have the same inode @@ -105,7 +109,7 @@ def get_inode(directory: DirectorySnapshot, full_path: str) -> int | Tuple[int, deleted.add(path) # find moved paths - moved: set[Tuple[str, str]] = set() + moved: set[tuple[str, str]] = set() for path in set(deleted): inode = ref.inode(path) new_path = snapshot.path(inode) @@ -165,22 +169,22 @@ def __repr__(self) -> str: ) @property - def files_created(self) -> List[str]: + def files_created(self) -> list[str]: """List of files that were created.""" return self._files_created @property - def files_deleted(self) -> List[str]: + def files_deleted(self) -> list[str]: """List of files that were deleted.""" return self._files_deleted @property - def files_modified(self) -> List[str]: + def files_modified(self) -> list[str]: """List of files that were modified.""" return self._files_modified @property - def files_moved(self) -> list[Tuple[str, str]]: + def files_moved(self) -> list[tuple[str, str]]: """List of files that were moved. Each event is a two-tuple the first item of which is the path @@ -189,12 +193,12 @@ def files_moved(self) -> list[Tuple[str, str]]: return self._files_moved @property - def dirs_modified(self) -> List[str]: + def dirs_modified(self) -> list[str]: """List of directories that were modified.""" return self._dirs_modified @property - def dirs_moved(self) -> List[tuple[str, str]]: + def dirs_moved(self) -> list[tuple[str, str]]: """List of directories that were moved. Each event is a two-tuple the first item of which is the path @@ -203,12 +207,12 @@ def dirs_moved(self) -> List[tuple[str, str]]: return self._dirs_moved @property - def dirs_deleted(self) -> List[str]: + def dirs_deleted(self) -> list[str]: """List of directories that were deleted.""" return self._dirs_deleted @property - def dirs_created(self) -> List[str]: + def dirs_created(self) -> list[str]: """List of directories that were created.""" return self._dirs_created @@ -313,7 +317,7 @@ def __init__( self.listdir = listdir self._stat_info: dict[str, os.stat_result] = {} - self._inode_to_path: dict[Tuple[int, int], str] = {} + self._inode_to_path: dict[tuple[int, int], str] = {} st = self.stat(path) self._stat_info[path] = st @@ -324,7 +328,7 @@ def __init__( self._inode_to_path[i] = p self._stat_info[p] = st - def walk(self, root: str) -> Iterator[Tuple[str, os.stat_result]]: + def walk(self, root: str) -> Iterator[tuple[str, os.stat_result]]: try: paths = [os.path.join(root, entry.name) for entry in self.listdir(root)] except OSError as e: @@ -355,11 +359,11 @@ 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]) -> Optional[str]: """Returns path for id. None if id is unknown to this snapshot.""" return self._inode_to_path.get(uid) - def inode(self, path: str) -> Tuple[int, int]: + def inode(self, path: str) -> tuple[int, int]: """Returns an id for path.""" st = self._stat_info[path] return (st.st_ino, st.st_dev) diff --git a/src/watchdog/utils/patterns.py b/src/watchdog/utils/patterns.py index f2db3713..b012d7ac 100644 --- a/src/watchdog/utils/patterns.py +++ b/src/watchdog/utils/patterns.py @@ -14,24 +14,35 @@ # - `PurePosixPath` is always case-sensitive. # Reference: https://docs.python.org/3/library/pathlib.html#pathlib.PurePath.match from pathlib import PurePosixPath, PureWindowsPath +from typing import TYPE_CHECKING +if TYPE_CHECKING: + from collections.abc import Iterator -def _match_path(path, included_patterns, excluded_patterns, case_sensitive): + +def _match_path(raw_path: str, included_patterns: set[str], excluded_patterns: set[str], case_sensitive: bool) -> bool: """Internal function same as :func:`match_path` but does not check arguments.""" + path: PurePosixPath | PureWindowsPath if case_sensitive: - path = PurePosixPath(path) + path = PurePosixPath(raw_path) else: included_patterns = {pattern.lower() for pattern in included_patterns} excluded_patterns = {pattern.lower() for pattern in excluded_patterns} - path = PureWindowsPath(path) + path = PureWindowsPath(raw_path) common_patterns = included_patterns & excluded_patterns if common_patterns: raise ValueError(f"conflicting patterns `{common_patterns}` included and excluded") + return any(path.match(p) for p in included_patterns) and not any(path.match(p) for p in excluded_patterns) -def filter_paths(paths, included_patterns=None, excluded_patterns=None, case_sensitive=True): +def filter_paths( + paths: list[str], + included_patterns: list[str] | None = None, + excluded_patterns: list[str] | None = None, + case_sensitive: bool = True, +) -> Iterator[str]: """Filters from a set of paths based on acceptable patterns and ignorable patterns. :param paths: @@ -58,7 +69,12 @@ def filter_paths(paths, included_patterns=None, excluded_patterns=None, case_sen yield path -def match_any_paths(paths, included_patterns=None, excluded_patterns=None, case_sensitive=True): +def match_any_paths( + paths: list[str], + included_patterns: list[str] | None = None, + excluded_patterns: list[str] | None = None, + case_sensitive: bool = True, +) -> bool: """Matches from a set of paths based on acceptable patterns and ignorable patterns. See ``filter_paths()`` for signature details. diff --git a/src/watchdog/utils/platform.py b/src/watchdog/utils/platform.py index 0f6b05a3..9d8ed576 100644 --- a/src/watchdog/utils/platform.py +++ b/src/watchdog/utils/platform.py @@ -24,7 +24,7 @@ PLATFORM_UNKNOWN = "unknown" -def get_platform_name(): +def get_platform_name() -> str: if sys.platform.startswith("win"): return PLATFORM_WINDOWS @@ -43,17 +43,17 @@ def get_platform_name(): __platform__ = get_platform_name() -def is_linux(): +def is_linux() -> bool: return __platform__ == PLATFORM_LINUX -def is_bsd(): +def is_bsd() -> bool: return __platform__ == PLATFORM_BSD -def is_darwin(): +def is_darwin() -> bool: return __platform__ == PLATFORM_DARWIN -def is_windows(): +def is_windows() -> bool: return __platform__ == PLATFORM_WINDOWS diff --git a/src/watchdog/watchmedo.py b/src/watchdog/watchmedo.py index 783d1e06..841fee35 100644 --- a/src/watchdog/watchmedo.py +++ b/src/watchdog/watchmedo.py @@ -31,13 +31,15 @@ from argparse import ArgumentParser, RawDescriptionHelpFormatter from io import StringIO from textwrap import dedent -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from watchdog.utils import WatchdogShutdown, load_class, platform from watchdog.version import VERSION_STRING if TYPE_CHECKING: - from watchdog.observers.api import BaseObserverSubclassCallable + from argparse import Namespace, _SubParsersAction + from typing import Callable + logging.basicConfig(level=logging.INFO) @@ -77,15 +79,21 @@ def _split_lines(self, text, width): subparsers = cli.add_subparsers(dest="top_command") command_parsers = {} +Argument = tuple[list[str], Any] + -def argument(*name_or_flags, **kwargs): +def argument(*name_or_flags: str, **kwargs: Any) -> Argument: """Convenience function to properly format arguments to pass to the command decorator. """ return list(name_or_flags), kwargs -def command(args=[], parent=subparsers, cmd_aliases=[]): +def command( + args: list[Argument], + parent: _SubParsersAction[ArgumentParser] = subparsers, + cmd_aliases: list[str] = [], +) -> Callable: """Decorator to define a new command in a sanity-preserving way. The function will be stored in the ``func`` variable when the parser parses arguments so that it can be called directly like so:: @@ -95,16 +103,16 @@ def command(args=[], parent=subparsers, cmd_aliases=[]): """ - def decorator(func): + def decorator(func: Callable) -> Callable: name = func.__name__.replace("_", "-") - desc = dedent(func.__doc__) - parser = parent.add_parser(name, description=desc, aliases=cmd_aliases, formatter_class=HelpFormatter) + desc = dedent(func.__doc__ or "") + parser = parent.add_parser(name, aliases=cmd_aliases, description=desc, formatter_class=HelpFormatter) command_parsers[name] = parser verbosity_group = parser.add_mutually_exclusive_group() verbosity_group.add_argument("-q", "--quiet", dest="verbosity", action="append_const", const=-1) verbosity_group.add_argument("-v", "--verbose", dest="verbosity", action="append_const", const=1) - for arg in args: - parser.add_argument(*arg[0], **arg[1]) + for name_or_flags, kwargs in args: + parser.add_argument(*name_or_flags, **kwargs) parser.set_defaults(func=func) return func @@ -198,8 +206,8 @@ def schedule_tricks(observer, tricks, pathname, recursive): """ for trick in tricks: for name, value in list(trick.items()): - TrickClass = load_class(name) - handler = TrickClass(**value) + trick_cls = load_class(name) + handler = trick_cls(**value) trick_pathname = getattr(handler, "source_directory", None) or pathname observer.schedule(handler, trick_pathname, recursive) @@ -252,7 +260,6 @@ def schedule_tricks(observer, tricks, pathname, recursive): ) 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: @@ -342,8 +349,8 @@ def tricks_generate_yaml(args): output = StringIO() for trick_path in args.trick_paths: - TrickClass = load_class(trick_path) - output.write(TrickClass.generate_yaml()) + trick_cls = load_class(trick_path) + output.write(trick_cls.generate_yaml()) content = output.getvalue() output.close() @@ -448,7 +455,6 @@ def log(args): 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: @@ -559,7 +565,6 @@ 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: @@ -672,7 +677,6 @@ def shell_command(args): ) 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: @@ -731,7 +735,7 @@ class LogLevelException(Exception): pass -def _get_log_level_from_args(args): +def _get_log_level_from_args(args: Namespace) -> str: verbosity = sum(args.verbosity or []) if verbosity < -1: raise LogLevelException("-q/--quiet may be specified only once.") @@ -740,7 +744,7 @@ def _get_log_level_from_args(args): return ["ERROR", "WARNING", "INFO", "DEBUG"][1 + verbosity] -def main(): +def main() -> int: """Entry-point function.""" args = cli.parse_args() if args.top_command is None: diff --git a/tests/test_observers_winapi.py b/tests/test_observers_winapi.py index 5c332d70..4bf0e15d 100644 --- a/tests/test_observers_winapi.py +++ b/tests/test_observers_winapi.py @@ -124,7 +124,7 @@ def test_root_deleted(event_queue, emitter): File "watchdog\observers\winapi.py", line 340, in read_directory_changes return _generate_observed_path_deleted_event() File "watchdog\observers\winapi.py", line 298, in _generate_observed_path_deleted_event - event = FILE_NOTIFY_INFORMATION(0, FILE_ACTION_DELETED_SELF, len(path), path.value) + event = FileNotifyInformation(0, FILE_ACTION_DELETED_SELF, len(path), path.value) TypeError: expected bytes, str found """ diff --git a/tests/utils.py b/tests/utils.py index 738bddee..91bee6b9 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -3,11 +3,11 @@ import dataclasses import os from queue import Empty, Queue -from typing import List, Optional, Tuple, Type, Union +from typing import Optional, Type, Union, Protocol from watchdog.events import FileSystemEvent from watchdog.observers.api import EventEmitter, ObservedWatch -from watchdog.utils import Protocol, platform +from watchdog.utils import platform Emitter: Type[EventEmitter] @@ -42,13 +42,13 @@ def __call__(self, expected_event: FileSystemEvent, timeout: float = ...) -> Non ... -TestEventQueue = Union["Queue[Tuple[FileSystemEvent, ObservedWatch]]"] +TestEventQueue = Queue[tuple[FileSystemEvent, ObservedWatch]] @dataclasses.dataclass() class Helper: tmp: str - emitters: List[EventEmitter] = dataclasses.field(default_factory=list) + emitters: list[EventEmitter] = dataclasses.field(default_factory=list) event_queue: TestEventQueue = dataclasses.field(default_factory=Queue) def joinpath(self, *args: str) -> str: diff --git a/tox.ini b/tox.ini index e74a35be..6bd6a5d3 100644 --- a/tox.ini +++ b/tox.ini @@ -1,6 +1,6 @@ [tox] envlist = - py3{8,9,10,11,12,13} + py3{9,10,11,12,13} pypy3 docs types