Repository: gorakhargosh/watchdog
Pull request #1011 (2023-10-06 21:07:16)

Add `DirectorySnapshotDiff.ContextManager`
A context manager that creates two directory snapshots and a diff object that represents the difference between the two snapshots.

```python
dir_snapshot_diff_context_manager = DirectorySnapshotDiff.ContextManager("some_path")
with dir_snapshot_diff_context_manager:
    # Do some things that change files...
    ...

print(dir_snapshot_diff_context_manager.diff.files_created)
print(dir_snapshot_diff_context_manager.diff.files_deleted)
```

BoboTiG: That is awesome :)
Would you mind rebasing your branch on master to benefit of your older PR (+ resolving conflicts ofc) 🙏🏻 ?
msabramo: > Would you mind rebasing your branch on master to benefit of your older PR (+ resolving conflicts ofc) 🙏🏻 ?

Would be happy to! Done!
BoboTiG: I just saw mypy issues. Are they related to the previous PR? 🤔 
msabramo: > I just saw mypy issues. Are they related to the previous PR? 🤔

See #1014...

Patch:
diff --git a/changelog.rst b/changelog.rst
index a18ba66..d6754dd 100644
--- a/changelog.rst
+++ b/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``
diff --git a/src/watchdog/observers/polling.py b/src/watchdog/observers/polling.py
index 466cd82..af74a6c 100644
--- a/src/watchdog/observers/polling.py
+++ b/src/watchdog/observers/polling.py
@@ -49,7 +49,7 @@
     FileMovedEvent,
 )
 from watchdog.observers.api import DEFAULT_EMITTER_TIMEOUT, DEFAULT_OBSERVER_TIMEOUT, BaseObserver, EventEmitter
-from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff
+from watchdog.utils.dirsnapshot import DirectorySnapshot, DirectorySnapshotDiff, EmptyDirectorySnapshot
 
 
 class PollingEmitter(EventEmitter):
@@ -68,7 +68,7 @@ def __init__(
         listdir=os.scandir,
     ):
         super().__init__(event_queue, watch, timeout, event_filter)
-        self._snapshot = None
+        self._snapshot: DirectorySnapshot = EmptyDirectorySnapshot()
         self._lock = threading.Lock()
         self._take_snapshot = lambda: DirectorySnapshot(
             self.watch.path, self.watch.is_recursive, stat=stat, listdir=listdir
diff --git a/src/watchdog/utils/dirsnapshot.py b/src/watchdog/utils/dirsnapshot.py
index 282487b..0a67c33 100644
--- a/src/watchdog/utils/dirsnapshot.py
+++ b/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:
     """
@@ -356,12 +423,15 @@ def __repr__(self) -> str:
         return str(self._stat_info)
 
 
-class EmptyDirectorySnapshot:
+class EmptyDirectorySnapshot(DirectorySnapshot):
     """Class to implement an empty snapshot. This is used together with
     DirectorySnapshot and DirectorySnapshotDiff in order to get all the files/folders
     in the directory as created.
     """
 
+    def __init__(self):
+        pass
+
     @staticmethod
     def path(_: Any) -> None:
         """Mock up method to return the path of the received inode. As the snapshot


Test patch:
diff --git a/tests/test_snapshot_diff.py b/tests/test_snapshot_diff.py
index 57f4d50..4059f88 100644
--- a/tests/test_snapshot_diff.py
+++ b/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"))