Add DirectorySnapshotDiff.ContextManager

changelog.rst

@@ -8,8 +8,8 @@ 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>`__)
+- [snapshot] Added ``DirectorySnapshotDiff.ContextManager`` (`#1011 <https://github.com/gorakhargosh/watchdog/pull/1011>`__)
 - [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``

src/watchdog/utils/dirsnapshot.py

@@ -220,6 +220,73 @@ def dirs_created(self) -> List[str]:
         """
         return self._dirs_created
 
+    class ContextManager:
+        """
+        Context manager that creates two directory snapshots and a
+        diff object that represents the difference between the two snapshots.
+
+        :param path:
+            The directory path for which a snapshot should be taken.
+        :type path:
+            ``str``
+        :param recursive:
+            ``True`` if the entire directory tree should be included in the
+            snapshot; ``False`` otherwise.
+        :type recursive:
+            ``bool``
+        :param stat:
+            Use custom stat function that returns a stat structure for path.
+            Currently only st_dev, st_ino, st_mode and st_mtime are needed.
+
+            A function taking a ``path`` as argument which will be called
+            for every entry in the directory tree.
+        :param listdir:
+            Use custom listdir function. For details see ``os.scandir``.
+        :param ignore_device:
+            A boolean indicating whether to ignore the device id or not.
+            By default, a file may be uniquely identified by a combination of its first
+            inode and its device id. The problem is that the device id may (or may not)
+            change between system boots. This problem would cause the DirectorySnapshotDiff
+            to think a file has been deleted and created again but it would be the
+            exact same file.
+            Set to True only if you are sure you will always use the same device.
+        :type ignore_device:
+            :class:`bool`
+        """
+
+        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,
+            ignore_device: bool = False,
+        ):
+            self.path = path
+            self.recursive = recursive
+            self.stat = stat
+            self.listdir = listdir
+            self.ignore_device = ignore_device
+
+        def __enter__(self):
+            self.pre_snapshot = self.get_snapshot()
+
+        def __exit__(self, *args):
+            self.post_snapshot = self.get_snapshot()
+            self.diff = DirectorySnapshotDiff(
+                self.pre_snapshot,
+                self.post_snapshot,
+                ignore_device=self.ignore_device,
+            )
+
+        def get_snapshot(self):
+            return DirectorySnapshot(
+                path=self.path,
+                recursive=self.recursive,
+                stat=self.stat,
+                listdir=self.listdir,
+            )
+
 
 class DirectorySnapshot:
     """

tests/test_snapshot_diff.py

@@ -55,6 +55,20 @@ def test_move_to(p):
     assert diff.files_created == [p("dir2", "b")]
 
 
+def test_move_to_with_context_manager(p):
+    mkdir(p("dir1"))
+    touch(p("dir1", "a"))
+    mkdir(p("dir2"))
+
+    dir1_cm = DirectorySnapshotDiff.ContextManager(p("dir1"))
+    dir2_cm = DirectorySnapshotDiff.ContextManager(p("dir2"))
+    with dir1_cm, dir2_cm:
+        mv(p("dir1", "a"), p("dir2", "b"))
+
+    assert dir1_cm.diff.files_deleted == [p("dir1", "a")]
+    assert dir2_cm.diff.files_created == [p("dir2", "b")]
+
+
 def test_move_from(p):
     mkdir(p("dir1"))
     mkdir(p("dir2"))