async_watcher

class watcher_fs.async_watcher.AsyncFileWatcher(path: str | List[str | Path], callback: Callable, trigger_type: TriggerType = TriggerType.PER_FILE, callback_extra: bool = False)

Bases: object

Manages watching a specific file path or pattern for changes (async version).

path

A string (glob pattern) or list of paths to watch.

callback

Async or sync function to call when changes are detected.

trigger_type

Type of trigger (PER_FILE or ANY_FILE).

callback_extra

If True, callback receives path and change type.

async dispatch_callback(change: Tuple[str, str] | List[Tuple[str, str]])

Execute the callback with the provided argument (async).

Parameters:

arg – Single (path, change_type) tuple or list of such tuples.

class watcher_fs.async_watcher.AsyncWatcher

Bases: object

Monitors file system changes and dispatches callbacks asynchronously.

watchers

List of AsyncFileWatcher instances tracking paths or patterns.

tracked_files

Dict mapping file paths to their last modification times.

file_to_watchers

Dict mapping file paths to indices of associated watchers.

last_run_time

Timestamp of the last check() call.

async check()

Check for file changes and dispatch callbacks asynchronously.

Updates tracked_files and file_to_watchers based on current file states. Detects added, modified, and deleted files, triggering callbacks accordingly. Sets last_run_time to the current time.

async register(paths: str | List[str | Path], callback: Callable, trigger_type: TriggerType = TriggerType.PER_FILE, callback_extra: bool = False)

Register a file path or pattern to watch for changes (async).

Parameters:

• paths – Glob pattern (str) or list of file paths to monitor.

• callback – Async or sync function to call when changes are detected.

• trigger_type – TriggerType enum (PER_FILE or ANY_FILE).

• callback_extra – If True, pass (path, change_type) to callback.

class watcher_fs.async_watcher.TriggerType(*values)

Bases: Enum

Enumeration defining the types of file change triggers for Watcher.

ANY_FILE = 'any_file'

PER_FILE = 'per_file'

async watcher_fs.async_watcher.main()