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): (source)

Undocumented

Parameters
fs:FileSystemCacheUndocumented
def add_watched_paths(self, paths): (source)

Undocumented

Parameters
paths:Iterable[str]Undocumented
def dump_file_data(self): (source)

Undocumented

Returns
Dict[str, Tuple[float, int, str]]Undocumented
def find_changed(self): (source)
Return paths that have changes since the last call, in the watched set.
Returns
AbstractSet[str]Undocumented
def remove_watched_paths(self, paths): (source)

Undocumented

Parameters
paths:Iterable[str]Undocumented
def set_file_data(self, path, data): (source)

Undocumented

Parameters
path:strUndocumented
data:FileDataUndocumented
def update_changed(self, remove, update): (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.

Parameters
remove:List[str]Undocumented
update:List[str]Undocumented
Returns
AbstractSet[str]Undocumented

Undocumented

def _find_changed(self, paths): (source)

Undocumented

Parameters
paths:Iterable[str]Undocumented
Returns
AbstractSet[str]Undocumented
def _update(self, path): (source)

Undocumented

Parameters
path:strUndocumented
_file_data: Dict[str, Optional[FileData]] = (source)

Undocumented

_paths: Set[str] = (source)

Undocumented