#22: Incorporate debouncer logic into notifykit (#23)

* #22 Inited event processor + file cache + started to integrate them into the watcher

* #22: Simplified the flow given the processor

* #22 Wait until events are received before returning to the python world

* #22 Lint py wrappers

* #22: Tried to wrap sleeping into the allow_threads

* #22: Added a new example

* 22: Fixing the termination of the notification process in the async mode

* #22 🔖 v0.0.3

* #22 linting

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "notifykit_lib"
-version = "0.0.2"
+version = "0.0.3"
 edition = "2021"
 license = "A toolkit for building applications watching filesystem changes"
 homepage = "https://github.com/roma-glushko/notifykit"
@@ -13,6 +13,8 @@ crossbeam-utils = "0.8.16"
 notify-debouncer-full = "0.3.1"
 notify = { version = "6.1.1"}
 pyo3 = {version = "0.20.0", features = ["extension-module", "abi3-py38"]}
+file-id = "0.2.1"
+walkdir = "2.4.0"
 
 [lib]
 name = "_notifykit_lib"

Makefile

@@ -60,3 +60,8 @@ clean:  # Clean all cache dirs
 	@rm -f .coverage
 	@rm -f .coverage.*
 	@rm -rf build
+	@rm -rf dist
+	@rm -rf sdist
+	@rm -rf site
+	@rm -rf .ruff_cache
+	@rm -rf target

examples/awatch_dir_recursively.py

@@ -0,0 +1,22 @@
+import asyncio
+import os
+from pathlib import Path
+
+from notifykit import Notifier
+
+
+async def watch(watched_dir: Path) -> None:
+    notifier = Notifier(debounce_ms=200, debug=True)
+
+    notifier.watch([watched_dir])
+
+    async for events in notifier:
+        # process your events
+        print(events)
+
+
+if __name__ == "__main__":
+    watched_dir = Path("./watched_dir")
+    os.makedirs(watched_dir, exist_ok=True)
+
+    asyncio.run(watch(watched_dir))

examples/watch_dir_recursively.py

@@ -5,11 +5,12 @@
 
 
 def watch(watched_dir: Path) -> None:
-    with Notifier(debounce_ms=200, debug=True) as notifier:
-        notifier.watch([watched_dir])
+    notifier = Notifier(debounce_ms=200, debug=True)
 
-        for event in notifier:
-            print(event)
+    notifier.watch([watched_dir])
+
+    for events in notifier:
+        print(events)
 
 
 if __name__ == "__main__":

notifykit/_notifier.py

@@ -1,6 +1,6 @@
-import asyncio
 from os import PathLike
-from typing import Sequence, Protocol, Optional, Any
+import anyio
+from typing import Sequence, Protocol, Optional, Any, List
 from notifykit._notifykit_lib import (
     WatcherWrapper,
     AccessEvent,
@@ -12,9 +12,15 @@
     RenameEvent,
 )
 
-Events = (
-    AccessEvent | CreateEvent | ModifyDataEvent | ModifyMetadataEvent | ModifyOtherEvent | DeleteEvent | RenameEvent
-)
+Event = AccessEvent | CreateEvent | ModifyDataEvent | ModifyMetadataEvent | ModifyOtherEvent | DeleteEvent | RenameEvent
+
+
+class AnyEvent(Protocol):
+    def is_set(self) -> bool:
+        ...
+
+    def set(self) -> None:
+        ...
 
 
 class NotifierT(Protocol):
@@ -44,51 +50,61 @@ class Notifier:
     Notifier collects filesystem events from the underlying watcher and expose them via sync/async API
     """
 
-    def __init__(self, debounce_ms: int, debounce_tick_rate_ms: Optional[int] = None, debug: bool = False) -> None:
+    def __init__(
+        self, debounce_ms: int = 200, tick_ms: int = 50, debug: bool = False, stop_event: Optional[AnyEvent] = None
+    ) -> None:
+        self._debounce_ms = debounce_ms
+        self._tick_ms = tick_ms
         self._debug = debug
 
-        self._watcher = WatcherWrapper(debounce_ms, debug, debounce_tick_rate_ms)
+        self._watcher = WatcherWrapper(debounce_ms, debug)
+        self._stop_event = stop_event if stop_event else anyio.Event()
 
     def watch(
-        self, paths: Sequence[PathLike[str]], recursive: bool = True, ignore_permission_errors: bool = False
+        self,
+        paths: Sequence[PathLike[str]],
+        recursive: bool = True,
+        ignore_permission_errors: bool = False,
     ) -> None:
         self._watcher.watch([str(path) for path in paths], recursive, ignore_permission_errors)
 
     def unwatch(self, paths: Sequence[str]) -> None:
         self._watcher.unwatch(list(paths))
 
-    def get(self) -> Optional[Events]:
-        return self._watcher.get()
+    def get(self) -> Optional[List[Event]]:
+        return self._watcher.get(self._tick_ms, self._stop_event)
 
-    def __enter__(self) -> "Notifier":
-        self._watcher.start()
-
-        return self
-
-    def __exit__(self, *args: Any, **kwargs: Any) -> None:
-        self._watcher.stop()
-
-    def __del__(self) -> None:
-        self._watcher.stop()
+    def set_stopping(self) -> None:
+        self._stop_event.set()
 
     def __aiter__(self) -> "Notifier":
         return self
 
     def __iter__(self) -> "Notifier":
         return self
 
-    def __next__(self) -> Events:
-        event = self._watcher.get()
+    def __next__(self) -> List[Event]:
+        events = self._watcher.get(self._tick_ms, self._stop_event)
 
-        if event is None:
+        if events is None:
             raise StopIteration
 
-        return event
+        return events
 
-    async def __anext__(self) -> Events:
-        event = await asyncio.to_thread(self._watcher.get)
+    async def __anext__(self) -> List[Event]:
+        CancelledError = anyio.get_cancelled_exc_class()
 
-        if event is None:
-            raise StopIteration
+        async with anyio.create_task_group() as tg:
+            try:
+                events = await anyio.to_thread.run_sync(self._watcher.get, self._tick_ms, self._stop_event)
+            except (CancelledError, KeyboardInterrupt):
+                self._stop_event.set()
+                # suppressing KeyboardInterrupt wouldn't stop it getting raised by the top level asyncio.run call
+                raise
+
+            tg.cancel_scope.cancel()
+
+            if events is None:
+                raise StopIteration
 
-        return event
+            return events

notifykit/_notifykit_lib.pyi

@@ -89,11 +89,8 @@ class WatcherWrapper:
         self,
         debounce_ms: int,
         debug: bool = False,
-        debounce_tick_rate_ms: Optional[int] = None,
     ) -> None: ...
     def watch(self, paths: List[str], recursive: bool = True, ignore_permission_errors: bool = False) -> None: ...
     def unwatch(self, paths: List[str]) -> None: ...
-    def get(self) -> Optional[Any]: ...
-    def close(self) -> None: ...
-    def start(self) -> None: ...
-    def stop(self) -> None: ...
+    def get(self, tick_ms: int, stop_event: Any) -> Optional[Any]: ...
+    def set_stopping(self) -> None: ...

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "notifykit"
-version = "0.0.2"
+version = "0.0.3"
 description = "A modern efficient Python toolkit for building applications that need to watch filesystem changes"
 authors = [
     {name = "Roman Glushko", email = "roman.glushko.m@gmail.com"},
@@ -13,6 +13,7 @@ classifiers = [
 ]
 
 dependencies = [
+    "anyio>=3.0.0",
 ]
 
 requires-python = '>=3.8'