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
| 27 | |
| 28 | |
| 29 | class 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 |
no outgoing calls
searching dependent graphs…