Watcher for file system changes among specific paths. All file system access is performed using FileSystemCache. We detect changed files by stat()ing them all and comparing hashes of potentially changed files. If a file has both size and mtime unmodified, the file is assumed to be u
| 16 | |
| 17 | |
| 18 | class FileSystemWatcher: |
| 19 | """Watcher for file system changes among specific paths. |
| 20 | |
| 21 | All file system access is performed using FileSystemCache. We |
| 22 | detect changed files by stat()ing them all and comparing hashes |
| 23 | of potentially changed files. If a file has both size and mtime |
| 24 | unmodified, the file is assumed to be unchanged. |
| 25 | |
| 26 | An important goal of this class is to make it easier to eventually |
| 27 | use file system events to detect file changes. |
| 28 | |
| 29 | Note: This class doesn't flush the file system cache. If you don't |
| 30 | manually flush it, changes won't be seen. |
| 31 | """ |
| 32 | |
| 33 | # TODO: Watching directories? |
| 34 | # TODO: Handle non-files |
| 35 | |
| 36 | def __init__(self, fs: FileSystemCache) -> None: |
| 37 | self.fs = fs |
| 38 | self._paths: set[str] = set() |
| 39 | self._file_data: dict[str, FileData | None] = {} |
| 40 | |
| 41 | def dump_file_data(self) -> dict[str, tuple[float, int, str]]: |
| 42 | return {k: v for k, v in self._file_data.items() if v is not None} |
| 43 | |
| 44 | def set_file_data(self, path: str, data: FileData) -> None: |
| 45 | self._file_data[path] = data |
| 46 | |
| 47 | def add_watched_paths(self, paths: Iterable[str]) -> None: |
| 48 | for path in paths: |
| 49 | if path not in self._paths: |
| 50 | # By storing None this path will get reported as changed by |
| 51 | # find_changed if it exists. |
| 52 | self._file_data[path] = None |
| 53 | self._paths |= set(paths) |
| 54 | |
| 55 | def remove_watched_paths(self, paths: Iterable[str]) -> None: |
| 56 | for path in paths: |
| 57 | if path in self._file_data: |
| 58 | del self._file_data[path] |
| 59 | self._paths -= set(paths) |
| 60 | |
| 61 | def _update(self, path: str, st: os.stat_result) -> None: |
| 62 | hash_digest = self.fs.hash_digest(path) |
| 63 | self._file_data[path] = FileData(st.st_mtime, st.st_size, hash_digest) |
| 64 | |
| 65 | def _find_changed(self, paths: Iterable[str]) -> AbstractSet[str]: |
| 66 | changed = set() |
| 67 | for path in paths: |
| 68 | old = self._file_data[path] |
| 69 | st = self.fs.stat_or_none(path) |
| 70 | if st is None: |
| 71 | if old is not None: |
| 72 | # File was deleted. |
| 73 | changed.add(path) |
| 74 | self._file_data[path] = None |
| 75 | else: |
no outgoing calls
no test coverage detected
searching dependent graphs…