Skip to content

Commit

Permalink
feat: Add typing to dirsnapshot (#1012)
Browse files Browse the repository at this point in the history
Before:

```
(.venv)
abramowi at Marcs-MacBook-Pro-3 in ~/Code/OpenSource/watchdog (master●)
$ mypy --disallow-untyped-defs src/watchdog/utils/dirsnapshot.py
src/watchdog/utils/dirsnapshot.py:82: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:88: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:93: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:141: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:144: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:162: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:167: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:172: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:177: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:187: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:194: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:204: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:211: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:241: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:258: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:290: error: Function is missing a return type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:296: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:302: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:307: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:310: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:313: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:316: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:331: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:340: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:343: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:354: error: Function is missing a type annotation  [no-untyped-def]
src/watchdog/utils/dirsnapshot.py:364: error: Function is missing a return type annotation  [no-untyped-def]
Found 27 errors in 1 file (checked 1 source file)
```

After:

```
(.venv)
abramowi at Marcs-MacBook-Pro-3 in ~/Code/OpenSource/watchdog (master●●)
$ mypy --disallow-untyped-defs src/watchdog/utils/dirsnapshot.py
Success: no issues found in 1 source file
```
  • Loading branch information
msabramo committed Oct 7, 2023
1 parent 75a3289 commit 5f9d93c
Show file tree
Hide file tree
Showing 2 changed files with 46 additions and 32 deletions.
4 changes: 3 additions & 1 deletion changelog.rst
Expand Up @@ -8,13 +8,15 @@ Changelog

2023-xx-xx • `full history <https://github.com/gorakhargosh/watchdog/compare/v3.0.0...HEAD>`__


- [snapshot] Add typing to ``dirsnapshot`` (`#1012 <https://github.com/gorakhargosh/watchdog/pull/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
~~~~~
Expand Down
74 changes: 43 additions & 31 deletions src/watchdog/utils/dirsnapshot.py
Expand Up @@ -51,6 +51,7 @@
import errno
import os
from stat import S_ISDIR
from typing import Any, Callable, Iterator, List, Optional, Tuple


class DirectorySnapshotDiff:
Expand Down Expand Up @@ -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
Expand All @@ -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)
Expand All @@ -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):
Expand All @@ -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})>"
Expand All @@ -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.
Expand All @@ -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.
Expand All @@ -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.
"""
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down Expand Up @@ -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.
Expand All @@ -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.
Expand All @@ -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)


Expand All @@ -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.
Expand All @@ -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.
Expand Down

0 comments on commit 5f9d93c

Please sign in to comment.