MCPcopy
hub / github.com/gorakhargosh/watchdog / ObservedWatch

Class ObservedWatch

src/watchdog/observers/api.py:29–83  ·  view source on GitHub ↗

An scheduled watch. :param path: Path string. :param recursive: ``True`` if watch is recursive; ``False`` otherwise. :param event_filter: Optional collection of :class:`watchdog.events.FileSystemEvent` to watch

Source from the content-addressed store, hash-verified

27
28
29class ObservedWatch:
30 """An scheduled watch.
31
32 :param path:
33 Path string.
34 :param recursive:
35 ``True`` if watch is recursive; ``False`` otherwise.
36 :param event_filter:
37 Optional collection of :class:`watchdog.events.FileSystemEvent` to watch
38 """
39
40 def __init__(self, path: str | Path, *, recursive: bool, event_filter: list[type[FileSystemEvent]] | None = None):
41 self._path = str(path) if isinstance(path, Path) else path
42 self._is_recursive = recursive
43 self._event_filter = frozenset(event_filter) if event_filter is not None else None
44
45 @property
46 def path(self) -> str:
47 """The path that this watch monitors."""
48 return self._path
49
50 @property
51 def is_recursive(self) -> bool:
52 """Determines whether subdirectories are watched for the path."""
53 return self._is_recursive
54
55 @property
56 def event_filter(self) -> frozenset[type[FileSystemEvent]] | None:
57 """Collection of event types watched for the path"""
58 return self._event_filter
59
60 @property
61 def key(self) -> tuple[str, bool, frozenset[type[FileSystemEvent]] | None]:
62 return self.path, self.is_recursive, self.event_filter
63
64 def __eq__(self, watch: object) -> bool:
65 if not isinstance(watch, ObservedWatch):
66 return NotImplemented
67 return self.key == watch.key
68
69 def __ne__(self, watch: object) -> bool:
70 if not isinstance(watch, ObservedWatch):
71 return NotImplemented
72 return self.key != watch.key
73
74 def __hash__(self) -> int:
75 return hash(self.key)
76
77 def __repr__(self) -> str:
78 if self.event_filter is not None:
79 event_filter_str = "|".join(sorted(_cls.__name__ for _cls in self.event_filter))
80 event_filter_str = f", event_filter={event_filter_str}"
81 else:
82 event_filter_str = ""
83 return f"<{type(self).__name__}: path={self.path!r}, is_recursive={self.is_recursive}{event_filter_str}>"
84
85
86# Observer classes

Callers 11

test_add_watch_twiceFunction · 0.90
start_watchingMethod · 0.90
emitterFunction · 0.90
emitterFunction · 0.90
test_observer__eq__Function · 0.90
test_observer__ne__Function · 0.90
test_observer__repr__Function · 0.90
test_event_emitterFunction · 0.90
test_event_dispatcherFunction · 0.90
scheduleMethod · 0.85

Calls

no outgoing calls

Tested by 9

test_add_watch_twiceFunction · 0.72
emitterFunction · 0.72
emitterFunction · 0.72
test_observer__eq__Function · 0.72
test_observer__ne__Function · 0.72
test_observer__repr__Function · 0.72
test_event_emitterFunction · 0.72
test_event_dispatcherFunction · 0.72

Used in the wild real call sites across dependent graphs

searching dependent graphs…