class documentation

class FileSystemWatcher: (source)

View In Hierarchy

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 unchanged. An important goal of this class is to make it easier to eventually use file system events to detect file changes. Note: This class doesn't flush the file system cache. If you don't manually flush it, changes won't be seen.

Method __init__ Undocumented
Method add_watched_paths Undocumented
Method dump_file_data Undocumented
Method find_changed Return paths that have changes since the last call, in the watched set.
Method remove_watched_paths Undocumented
Method set_file_data Undocumented
Method update_changed Alternative to find_changed() given explicit changes.
Instance Variable fs Undocumented
Method _find_changed Undocumented
Method _update Undocumented
Instance Variable _file_data Undocumented
Instance Variable _paths Undocumented
def __init__(self, fs: FileSystemCache): (source)

Undocumented

def add_watched_paths(self, paths: Iterable[str]): (source)

Undocumented

def dump_file_data(self) -> dict[str, tuple[float, int, str]]: (source)

Undocumented

def find_changed(self) -> AbstractSet[str]: (source)

Return paths that have changes since the last call, in the watched set.

def remove_watched_paths(self, paths: Iterable[str]): (source)

Undocumented

def set_file_data(self, path: str, data: FileData): (source)

Undocumented

def update_changed(self, remove: list[str], update: list[str]) -> AbstractSet[str]: (source)

Alternative to find_changed() given explicit changes. This only calls self.fs.stat() on added or updated files, not on all files. It believes all other files are unchanged! Implies add_watched_paths() for add and update, and remove_watched_paths() for remove.

Undocumented

def _find_changed(self, paths: Iterable[str]) -> AbstractSet[str]: (source)

Undocumented

def _update(self, path: str): (source)

Undocumented

Undocumented

Undocumented