diff --git a/changelog.rst b/changelog.rst index 5a2d944e..a18ba665 100644 --- a/changelog.rst +++ b/changelog.rst @@ -8,13 +8,15 @@ Changelog 2023-xx-xx • `full history `__ + +- [snapshot] Add typing to ``dirsnapshot`` (`#1012 `__) - [events] ``FileSystemEvent``, and subclasses, are now ``dataclass``es, and their ``repr()`` has changed - [windows] ``WinAPINativeEvent`` is now a ``dataclass``, and its ``repr()`` has changed - [events] Log ``FileOpenedEvent``, and ``FileClosedEvent``, events in ``LoggingEventHandler`` - [tests] Improve ``FileSystemEvent`` coverage - [watchmedo] Log all events in ``LoggerTrick`` - [windows] The ``observers.read_directory_changes.WATCHDOG_TRAVERSE_MOVED_DIR_DELAY`` hack was removed. The constant will be kept to prevent breaking other softwares. -- Thanks to our beloved contributors: @BoboTiG +- Thanks to our beloved contributors: @BoboTiG, @msabramo 3.0.0 ~~~~~ diff --git a/src/watchdog/utils/dirsnapshot.py b/src/watchdog/utils/dirsnapshot.py index c1ff0331..282487ba 100644 --- a/src/watchdog/utils/dirsnapshot.py +++ b/src/watchdog/utils/dirsnapshot.py @@ -51,6 +51,7 @@ import errno import os from stat import S_ISDIR +from typing import Any, Callable, Iterator, List, Optional, Tuple class DirectorySnapshotDiff: @@ -79,18 +80,23 @@ class DirectorySnapshotDiff: :class:`bool` """ - def __init__(self, ref, snapshot, ignore_device=False): + def __init__( + self, + ref: DirectorySnapshot, + snapshot: DirectorySnapshot, + ignore_device: bool = False, + ): created = snapshot.paths - ref.paths deleted = ref.paths - snapshot.paths if ignore_device: - def get_inode(directory, full_path): + def get_inode(directory: DirectorySnapshot, full_path: str) -> int | Tuple[int, int]: return directory.inode(full_path)[0] else: - def get_inode(directory, full_path): + 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 @@ -100,7 +106,7 @@ def get_inode(directory, full_path): deleted.add(path) # find moved paths - moved = set() + moved: set[Tuple[str, str]] = set() for path in set(deleted): inode = ref.inode(path) new_path = snapshot.path(inode) @@ -118,7 +124,7 @@ def get_inode(directory, full_path): # find modified paths # first check paths that have not moved - modified = set() + modified: set[str] = set() for path in ref.paths & snapshot.paths: if get_inode(ref, path) == get_inode(snapshot, path): if ref.mtime(path) != snapshot.mtime(path) or ref.size(path) != snapshot.size(path): @@ -138,10 +144,10 @@ def get_inode(directory, full_path): self._files_modified = list(modified - set(self._dirs_modified)) self._files_moved = list(moved - set(self._dirs_moved)) - def __str__(self): + def __str__(self) -> str: return self.__repr__() - def __repr__(self): + def __repr__(self) -> str: fmt = ( "<{0} files(created={1}, deleted={2}, modified={3}, moved={4})," " folders(created={5}, deleted={6}, modified={7}, moved={8})>" @@ -159,22 +165,22 @@ def __repr__(self): ) @property - def files_created(self): + def files_created(self) -> List[str]: """List of files that were created.""" return self._files_created @property - def files_deleted(self): + def files_deleted(self) -> List[str]: """List of files that were deleted.""" return self._files_deleted @property - def files_modified(self): + def files_modified(self) -> List[str]: """List of files that were modified.""" return self._files_modified @property - def files_moved(self): + def files_moved(self) -> list[Tuple[str, str]]: """ List of files that were moved. @@ -184,14 +190,14 @@ def files_moved(self): return self._files_moved @property - def dirs_modified(self): + def dirs_modified(self) -> List[str]: """ List of directories that were modified. """ return self._dirs_modified @property - def dirs_moved(self): + def dirs_moved(self) -> List[tuple[str, str]]: """ List of directories that were moved. @@ -201,14 +207,14 @@ def dirs_moved(self): return self._dirs_moved @property - def dirs_deleted(self): + def dirs_deleted(self) -> List[str]: """ List of directories that were deleted. """ return self._dirs_deleted @property - def dirs_created(self): + def dirs_created(self) -> List[str]: """ List of directories that were created. """ @@ -238,13 +244,19 @@ class DirectorySnapshot: Use custom listdir function. For details see ``os.scandir``. """ - def __init__(self, path, recursive=True, stat=os.stat, listdir=os.scandir): + def __init__( + self, + path: str, + recursive: bool = True, + stat: Callable[[str], os.stat_result] = os.stat, + listdir: Callable[[Optional[str]], Iterator[os.DirEntry]] = os.scandir, + ): self.recursive = recursive self.stat = stat self.listdir = listdir - self._stat_info = {} - self._inode_to_path = {} + self._stat_info: dict[str, os.stat_result] = {} + self._inode_to_path: dict[Tuple[int, int], str] = {} st = self.stat(path) self._stat_info[path] = st @@ -255,7 +267,7 @@ def __init__(self, path, recursive=True, stat=os.stat, listdir=os.scandir): self._inode_to_path[i] = p self._stat_info[p] = st - def walk(self, root): + 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: @@ -287,33 +299,33 @@ def walk(self, root): pass @property - def paths(self): + def paths(self) -> set[str]: """ Set of file/directory paths in the snapshot. """ return set(self._stat_info.keys()) - def path(self, id): + def path(self, id: Tuple[int, int]) -> Optional[str]: """ Returns path for id. None if id is unknown to this snapshot. """ return self._inode_to_path.get(id) - def inode(self, path): + 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) - def isdir(self, path): + def isdir(self, path: str) -> bool: return S_ISDIR(self._stat_info[path].st_mode) - def mtime(self, path): + def mtime(self, path: str) -> float: return self._stat_info[path].st_mtime - def size(self, path): + def size(self, path: str) -> int: return self._stat_info[path].st_size - def stat_info(self, path): + def stat_info(self, path: str) -> os.stat_result: """ Returns a stat information object for the specified path from the snapshot. @@ -328,7 +340,7 @@ def stat_info(self, path): """ return self._stat_info[path] - def __sub__(self, previous_dirsnap): + def __sub__(self, previous_dirsnap: DirectorySnapshot) -> DirectorySnapshotDiff: """Allow subtracting a DirectorySnapshot object instance from another. @@ -337,10 +349,10 @@ def __sub__(self, previous_dirsnap): """ return DirectorySnapshotDiff(previous_dirsnap, self) - def __str__(self): + def __str__(self) -> str: return self.__repr__() - def __repr__(self): + def __repr__(self) -> str: return str(self._stat_info) @@ -351,7 +363,7 @@ class EmptyDirectorySnapshot: """ @staticmethod - def path(_): + def path(_: Any) -> None: """Mock up method to return the path of the received inode. As the snapshot is intended to be empty, it always returns None. @@ -361,7 +373,7 @@ def path(_): return None @property - def paths(self): + def paths(self) -> set: """Mock up method to return a set of file/directory paths in the snapshot. As the snapshot is intended to be empty, it always returns an empty set.