Coverage for core / src / sensorkit / common / filewatch.py: 85%
238 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-09-02 00:03 +0000
1# SPDX-License-Identifier: Apache-2.0
2"""Async-friendly filesystem watching over watchdog.
4Wraps watchdog's threaded, callback-based API in an asyncio-native interface. `watch_dir`
5watches a directory for the duration of a block; `wait_for_file` suspends until a single
6file appears. Every directory watch in the process shares one Observer.
8Caveats:
9- The watched directory (and, for `wait_for_file`, the file's parent directory) must already
10 exist; scheduling a watch on a missing path raises.
11- A directory gets one physical watch, whose recursion is fixed by its first subscriber. A
12 non-recursive consumer can share a recursive watch, but requesting `recursive=True` for a
13 directory already being watched non-recursively raises ValueError.
14- A temp-then-rename write surfaces as `MOVED` (path = destination), not `CREATED`;
15 include `MOVED` in `kinds` when watching for "a file became ready".
16- `existing=True` reports already-present entries with kind `EXISTING`, scanned after
17 subscribing so no live event is missed. A present entry may therefore appear both via
18 the scan and a concurrent live event; consumers should be idempotent.
19- Queues are unbounded by default (lossless); pass `max_queue` to bound them, in which
20 case the newest event is dropped and logged on overflow.
21- `wait_for_file` polls rather than using a native watcher, so it also sees remote writes
22 on network mounts, at a detection latency of up to `poll_interval`.
23"""
25from __future__ import annotations
27import asyncio
28import contextlib
29import enum
30import functools
31import os
32import pathlib
33import sys
34import threading
35from collections.abc import AsyncGenerator, AsyncIterator, Callable, Collection
36from dataclasses import dataclass
37from queue import Queue
39from loguru import logger
40from watchdog.events import FileSystemEvent, FileSystemEventHandler
41from watchdog.observers import Observer
42from watchdog.observers.api import BaseObserver, ObservedWatch
43from watchdog.observers.polling import PollingObserver
45__all__ = [
46 "FileEvent",
47 "FileEventKind",
48 "watch_dir",
49 "wait_for_file",
50]
53class FileEventKind(enum.Enum):
54 """The kind of filesystem change a `FileEvent` represents.
56 The live-event values mirror watchdog's `EVENT_TYPE_*` strings, so a watchdog event
57 maps straight back via `FileEventKind(event.event_type)`. `EXISTING` has no watchdog
58 counterpart: it marks an entry surfaced by an initial existing-file scan, rather than
59 a change observed live.
60 """
62 CREATED = "created"
63 MODIFIED = "modified"
64 MOVED = "moved"
65 DELETED = "deleted"
66 EXISTING = "existing"
69@dataclass(frozen=True, slots=True)
70class FileEvent:
71 """An observed filesystem change.
73 Attributes:
74 kind: What happened to the path.
75 path: The affected path. For `MOVED` events this is the *destination*.
76 src_path: The original path for `MOVED` events; `None` otherwise.
77 is_directory: Whether the affected path is a directory.
78 """
80 kind: FileEventKind
81 path: pathlib.Path
82 src_path: pathlib.Path | None
83 is_directory: bool
86def to_file_event(event: FileSystemEvent) -> FileEvent | None:
87 """Translate a watchdog event into a `FileEvent`, or `None` if we don't model it.
89 Returns `None` for event types without a `FileEventKind` (e.g. opened/closed).
90 """
91 try:
92 kind = FileEventKind(event.event_type)
93 except ValueError:
94 return None
96 if kind is FileEventKind.MOVED:
97 return FileEvent(
98 kind=kind,
99 path=pathlib.Path(os.fsdecode(event.dest_path)),
100 src_path=pathlib.Path(os.fsdecode(event.src_path)),
101 is_directory=event.is_directory,
102 )
104 return FileEvent(
105 kind=kind,
106 path=pathlib.Path(os.fsdecode(event.src_path)),
107 src_path=None,
108 is_directory=event.is_directory,
109 )
112class Subscriber:
113 """A single async consumer of events from one directory watch.
115 The watchdog observer thread calls `deliver` (off the event loop); it filters via
116 *predicate* and marshals matching events onto *queue* via the subscriber's loop.
117 """
119 def __init__(
120 self,
121 *,
122 loop: asyncio.AbstractEventLoop,
123 queue: asyncio.Queue[FileEvent],
124 predicate: Callable[[FileEvent], bool],
125 real_dir: str,
126 ):
127 self._loop = loop
128 self._queue = queue
129 self._predicate = predicate
130 self._real_dir = real_dir
132 def deliver(self, event: FileEvent) -> None:
133 """Called from the observer thread; hand a matching event to the consumer loop."""
134 if not self._predicate(event):
135 return
137 try:
138 self._loop.call_soon_threadsafe(self._put, event)
139 except RuntimeError:
140 # The consumer's event loop is closed; nothing to deliver to.
141 pass
143 def _put(self, event: FileEvent) -> None:
144 try:
145 self._queue.put_nowait(event)
146 except asyncio.QueueFull:
147 logger.warning(
148 f"filewatch queue full for {self._real_dir!r}, dropping event for {event.path}"
149 )
150 except asyncio.QueueShutDown:
151 pass
154class DispatchHandler(FileSystemEventHandler):
155 """The single watchdog handler per directory; filters and bridges events to subscribers.
157 watchdog already shares one emitter per directory across this handler -- we are not
158 reimplementing that. This handler's job is the async bridge (onto each subscriber's
159 event loop) and per-subscriber filtering, which watchdog does not provide.
160 """
162 def __init__(self, watch: DirWatch):
163 self._watch = watch
165 def on_any_event(self, event: FileSystemEvent) -> None:
166 file_event = to_file_event(event)
167 if file_event is None:
168 return
170 # Convert once, then snapshot under the lock and deliver outside it. Delivery is a
171 # non-blocking call_soon_threadsafe, but we avoid iterating the live set.
172 with self._watch.lock:
173 subscribers = tuple(self._watch.subscribers)
175 for subscriber in subscribers:
176 subscriber.deliver(file_event)
179class DirWatch:
180 """One physical watch on a directory, shared by its subscribers.
182 `recursive` is set by the first subscriber. A non-recursive consumer may share a
183 recursive watch (filtered); a recursive consumer cannot share a non-recursive watch and
184 is rejected (re-creating the watch would drop events on existing subscribers).
185 """
187 def __init__(self):
188 self.lock = threading.Lock()
189 self.subscribers: set[Subscriber] = set()
190 self.handler = DispatchHandler(self)
191 self.observed_watch: ObservedWatch | None = None
192 self.recursive = False
195class WatchManager:
196 """Process-wide registry funneling every watch through one shared Observer.
198 Thread-safety: `subscribe`/`unsubscribe` may be called from different event-loop
199 threads (e.g. tests).
200 """
202 def __init__(self):
203 self._lock = threading.Lock()
204 self._observer: BaseObserver | None = None
205 self._watches: dict[str, DirWatch] = {}
206 self._pending: Queue[tuple[str, ObservedWatch]] = Queue()
207 self._reaper: threading.Thread | None = None
209 def subscribe(self, real_dir: str, subscriber: Subscriber, *, recursive: bool) -> None:
210 with self._lock:
211 watch = self._watches.get(real_dir)
213 if watch is None:
214 watch = DirWatch()
215 observer = self._ensure_observer_locked()
216 watch.recursive = recursive
217 watch.observed_watch = observer.schedule(
218 watch.handler, real_dir, recursive=recursive
219 )
220 self._watches[real_dir] = watch
221 elif recursive and not watch.recursive:
222 # The directory is already watched non-recursively. A recursive consumer
223 # cannot be served from that watch, and re-creating it as recursive would drop
224 # events on the existing subscribers during the gap -- so reject it. (The
225 # reverse, a non-recursive consumer of a recursive watch, is fine: it filters.)
226 raise ValueError(
227 f"{real_dir!r} is already watched non-recursively; it cannot also be "
228 f"watched recursively (one shared emitter per directory)"
229 )
231 with watch.lock:
232 watch.subscribers.add(subscriber)
234 def unsubscribe(self, real_dir: str, subscriber: Subscriber) -> None:
235 """Drop *subscriber*, queueing the physical unschedule if it was the last one.
237 Non-blocking, and safe to call from an event loop thread.
238 """
239 with self._lock:
240 watch = self._watches.get(real_dir)
241 if watch is None:
242 return
244 with watch.lock:
245 watch.subscribers.discard(subscriber)
246 if watch.subscribers:
247 return
249 del self._watches[real_dir]
251 if self._observer is None or watch.observed_watch is None:
252 return
254 self._pending.put((real_dir, watch.observed_watch))
255 self._ensure_reaper_locked()
257 def _reap(self) -> None:
258 """Unschedule watches queued by `unsubscribe`, one at a time. Runs on its own thread."""
259 # This exists to keep the unschedule off the caller's thread: it joins the watch's
260 # emitter with no timeout, and a PollingEmitter only notices the stop between
261 # passes, so the join can wait out a full walk of the tree. It holds the observer
262 # lock throughout, stalling dispatch for every other watch as well.
263 while True:
264 real_dir, observed_watch = self._pending.get()
266 try:
267 with self._lock:
268 self._unschedule_locked(real_dir, observed_watch)
269 except Exception:
270 # The reaper is process-wide: letting it die would leak every later watch.
271 logger.exception(f"filewatch failed to unschedule watch on {real_dir!r}")
272 finally:
273 self._pending.task_done()
275 def _unschedule_locked(self, real_dir: str, observed_watch: ObservedWatch) -> None:
276 observer = self._observer
278 if observer is None:
279 return
281 # Make sure a watch hasn't been re-added. An ObservedWatch compares by path and
282 # recursion and scheduling an equal one reuses the emitter, so unscheduling here
283 # would tear down the new subscriber's watch. One re-added with the opposite
284 # recursion is a distinct emitter and must still be reaped.
285 live = self._watches.get(real_dir)
287 if live is not None and live.observed_watch == observed_watch:
288 return
290 try:
291 observer.unschedule(observed_watch)
292 except KeyError:
293 # Already gone, e.g. via `unschedule_all`.
294 pass
296 def _ensure_observer_locked(self) -> BaseObserver:
297 observer = self._observer
299 if observer is None:
300 # One Observer process-wide: on macOS, two watches on the same directory in
301 # separate Observer instances collide, whereas one Observer keeps a single
302 # emitter per directory and dispatches it to every handler.
303 observer = Observer()
304 observer.start()
305 self._observer = observer
307 return observer
309 def _ensure_reaper_locked(self) -> None:
310 if self._reaper is None:
311 self._reaper = threading.Thread(
312 target=self._reap, name="filewatch-reaper", daemon=True
313 )
314 self._reaper.start()
316 def _reset(self) -> None:
317 """Drop all watches, leaving the shared Observer running. Intended for tests."""
318 # Let queued teardowns finish first, so they cannot fire against a later test's
319 # watches. Blocking here is fine: `_reset` runs off the event loop.
320 self._pending.join()
322 with self._lock:
323 observer = self._observer
324 self._watches.clear()
326 if observer is not None:
327 observer.unschedule_all()
330def patch_windows_emitter_handle_close() -> None:
331 """Make watchdog's Windows emitter close its directory handle at most once.
333 Through watchdog 6.0.0 the Windows emitter closes its directory handle from whichever
334 thread stops it and never clears the attribute, and `on_thread_stop` runs more than
335 once per emitter during ordinary teardown. The handle is therefore closed twice, and
336 between the two closes Windows is free to hand that handle value to something else --
337 so the second close destroys an unrelated object. When the value has been reused for
338 one of CPython's parking-lot semaphores, the interpreter dies with "Fatal Python
339 error: _PySemaphore_Wakeup: parking_lot: ReleaseSemaphore failed". Under pytest the
340 message is swallowed by output capture, leaving only a silent non-zero exit.
342 Clearing the attribute before closing makes the second call a no-op. Upstream fixes
343 this by rewriting the emitter around DirectoryChangeReader, which keeps the handle on
344 the thread that owns it; this patch detects that version and does nothing, and the
345 whole function can be dropped once it ships.
347 See https://github.com/gorakhargosh/watchdog/issues/1132.
348 """
349 from watchdog.observers import winapi
350 from watchdog.observers.read_directory_changes import WindowsApiEmitter
352 if hasattr(winapi, "DirectoryChangeReader"):
353 return
355 def on_thread_stop(self: WindowsApiEmitter) -> None:
356 whandle = self._whandle
357 if whandle:
358 self._whandle = None
359 winapi.close_directory_handle(whandle)
361 WindowsApiEmitter.on_thread_stop = on_thread_stop
364if sys.platform == "win32":
365 patch_windows_emitter_handle_close()
367manager = WatchManager()
370def event_matches(
371 event: FileEvent,
372 *,
373 kinds: frozenset[FileEventKind] | None,
374 real_dir: pathlib.Path,
375 recursive: bool,
376) -> bool:
377 """Whether *event* should be reported to a subscriber watching *real_dir*.
379 Filters by kind, drops events on the watched directory itself, and -- for a
380 non-recursive consumer -- drops anything below its immediate children, which is what
381 lets such a consumer share a recursive physical watch.
382 """
383 if kinds is not None and event.kind not in kinds:
384 return False
385 if event.path == real_dir:
386 return False
387 return recursive or event.path.parent == real_dir
390async def event_stream(
391 *,
392 queue: asyncio.Queue[FileEvent],
393 existing_events: list[FileEvent],
394 existing_done: asyncio.Event | None,
395) -> AsyncGenerator[FileEvent, None]:
396 """Yield the buffered initial scan, then live events."""
397 # Deliberately owns no cleanup: an async generator closed or dropped before its first
398 # `__anext__` never runs its body, so unsubscribing here would be skipped exactly when
399 # nothing was consumed. The enclosing `watch_dir` block does it instead.
400 for event in existing_events:
401 yield event
403 # The buffered scan is drained; a sequential consumer has processed it all by now.
404 if existing_done is not None:
405 existing_done.set()
407 while True:
408 yield await queue.get()
409 queue.task_done()
412def scan_existing(real_dir: str, recursive: bool) -> list[FileEvent]:
413 """Synthesize `EXISTING` events for entries already present under *real_dir*."""
414 base = pathlib.Path(real_dir)
415 entries = base.rglob("*") if recursive else base.iterdir()
417 return [
418 FileEvent(
419 kind=FileEventKind.EXISTING,
420 path=path,
421 src_path=None,
422 is_directory=path.is_dir(),
423 )
424 for path in entries
425 ]
428@contextlib.asynccontextmanager
429async def watch_dir(
430 directory: str | os.PathLike[str],
431 *,
432 recursive: bool = True,
433 kinds: Collection[FileEventKind] | None = None,
434 existing: bool = False,
435 existing_done: asyncio.Event | None = None,
436 max_queue: int = 0,
437) -> AsyncIterator[AsyncGenerator[FileEvent, None]]:
438 """Watch *directory* for the duration of the block, yielding a stream of its events.
440 Entering the block subscribes -- the watch is armed and, with `existing=True`, the
441 initial scan is complete by the time the body starts, so a caller may scan the
442 directory itself without racing live writes. Leaving it unsubscribes, however the
443 block exits.
445 Use it directly, or compose several watches with `contextlib.AsyncExitStack`:
447 async with watch_dir(root, kinds=(FileEventKind.CREATED,)) as events:
448 async for event in events:
449 ...
451 Args:
452 directory: Directory to watch. Must already exist.
453 recursive: When `False`, only events whose parent is *directory* itself are
454 reported. Raises ValueError if `True` and the directory is already watched
455 non-recursively.
456 kinds: If given, only report events of these kinds; otherwise report all. To
457 include the initial scan when `existing=True`, include `EXISTING`.
458 existing: When `True`, the stream first yields an `EXISTING` event for each entry
459 already present (scanned after subscribing, so no live event is missed). A present
460 entry may therefore also be reported by a concurrent live event.
461 existing_done: If given, set once the stream has yielded the last buffered
462 `EXISTING` event and is about to await live events. For a consumer that processes
463 each event before requesting the next (an `async for` that awaits in its body),
464 this fires right after the initial scan has been fully *processed* -- the signal a
465 caller needs to mark an initial listing complete. Fires on first iteration when
466 `existing=False` (the initial scan is empty). Intended for use with `existing=True`.
467 max_queue: Maximum number of buffered events (0 means unbounded). On overflow the
468 newest event is dropped and logged.
470 Yields:
471 An async generator of matching `FileEvent`s, live until the block exits.
472 """
473 real_dir = os.path.realpath(directory)
474 loop = asyncio.get_running_loop()
475 queue: asyncio.Queue[FileEvent] = asyncio.Queue(maxsize=max_queue)
477 predicate = functools.partial(
478 event_matches,
479 kinds=frozenset(kinds) if kinds is not None else None,
480 real_dir=pathlib.Path(real_dir),
481 recursive=recursive,
482 )
483 subscriber = Subscriber(loop=loop, queue=queue, predicate=predicate, real_dir=real_dir)
485 def subscribe_and_scan() -> list[FileEvent]:
486 # Subscribing blocks off the loop: it opens a directory handle and starts watchdog's
487 # threads, and a recursive inotify watch walks the tree adding one watch per
488 # directory -- slow enough on a deep tree or a stalled network mount to matter.
489 manager.subscribe(real_dir, subscriber, recursive=recursive)
491 # The watch is now live: matching events are delivered to `queue` and cannot be
492 # missed. Only now scan for pre-existing entries, and undo the subscription if that
493 # fails, since the block whose exit would otherwise clean up is never entered.
494 try:
495 return scan_existing(real_dir, recursive) if existing else []
496 except BaseException:
497 manager.unsubscribe(real_dir, subscriber)
498 raise
500 def drop_orphaned_subscription(task: asyncio.Task[list[FileEvent]]) -> None:
501 if not task.cancelled() and task.exception() is None:
502 manager.unsubscribe(real_dir, subscriber)
504 # Shield the worker rather than abandoning it on cancellation: a thread cannot be
505 # interrupted, so a subscription that lands after we stop waiting would never be taken
506 # back out. The initial listing is complete by the time this coroutine returns.
507 subscribing = asyncio.create_task(asyncio.to_thread(subscribe_and_scan))
509 try:
510 existing_events = [e for e in await asyncio.shield(subscribing) if predicate(e)]
511 except BaseException:
512 subscribing.add_done_callback(drop_orphaned_subscription)
513 raise
515 stream = event_stream(
516 queue=queue,
517 existing_events=existing_events,
518 existing_done=existing_done,
519 )
521 try:
522 yield stream
523 finally:
524 # Synchronous by design: this also runs when the block is left by cancellation or
525 # during interpreter shutdown, where awaiting can raise or never resume and the
526 # watch would leak for the life of the process. `unsubscribe` is non-blocking, so
527 # the emitter join it defers costs the caller nothing here.
528 manager.unsubscribe(real_dir, subscriber)
529 await stream.aclose()
532class WaitFileHandler(FileSystemEventHandler):
533 """Sets an asyncio.Event when a file of the target name appears in the watched directory.
535 Used by `wait_for_file`. Runs on the PollingObserver thread, so it marshals the signal
536 back to the waiting loop via `call_soon_threadsafe`.
537 """
539 def __init__(self, name: str, loop: asyncio.AbstractEventLoop, appeared: asyncio.Event):
540 self._name = name
541 self._loop = loop
542 self._appeared = appeared
544 def _signal(self, raw_path: bytes | str) -> None:
545 if os.path.basename(os.fsdecode(raw_path)) != self._name:
546 return
547 try:
548 self._loop.call_soon_threadsafe(self._appeared.set)
549 except RuntimeError:
550 # The waiting loop is gone; nothing to signal.
551 pass
553 def on_created(self, event: FileSystemEvent) -> None:
554 self._signal(event.src_path)
556 def on_moved(self, event: FileSystemEvent) -> None:
557 self._signal(event.dest_path)
560async def wait_for_file(path: str | os.PathLike[str], *, poll_interval: float = 1.0) -> None:
561 """Suspend until *path* exists, polling its parent directory.
563 Returns immediately if the file already exists. The file's parent directory must exist.
564 Uses a dedicated, short-lived PollingObserver rather than the shared Observer: polling
565 works on any filesystem (including network mounts, where native watchers miss remote
566 writes) and cannot collide with the shared Observer on macOS. Detection latency is up to
567 `poll_interval` seconds.
568 """
569 target = pathlib.Path(path)
570 if await asyncio.to_thread(target.exists):
571 return
573 loop = asyncio.get_running_loop()
574 appeared = asyncio.Event()
575 observer = PollingObserver(timeout=poll_interval)
576 observer.schedule(WaitFileHandler(target.name, loop, appeared), str(target.parent))
577 observer.start()
579 try:
580 # Re-check: the file may have appeared between the check above and the observer start.
581 if not await asyncio.to_thread(target.exists):
582 await appeared.wait()
583 finally:
584 observer.stop()
585 await asyncio.to_thread(observer.join)