Watch

watch

watch(*paths: Union[Path, str], watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(), debounce: int = 1600, step: int = 50, stop_event: Optional[AbstractEvent] = None, rust_timeout: int = 5000, yield_on_timeout: bool = False, debug: Optional[bool] = None, raise_interrupt: bool = True, force_polling: Optional[bool] = None, poll_delay_ms: int = 300, recursive: bool = True, ignore_permission_denied: Optional[bool] = None) -> Generator[Set[FileChange], None, None]

Watch one or more paths and yield a set of changes whenever files change.

The paths watched can be directories or files, directories are watched recursively - changes in subdirectories are also detected.

Force polling

Notify will fall back to file polling if it can't use file system notifications, but we also force Notify to use polling if the force_polling argument is True; if force_polling is unset (or None), we enable force polling thus:

• if the WATCHFILES_FORCE_POLLING environment variable exists and is not empty:
  • if the value is false, disable or disabled, force polling is disabled
  • otherwise, force polling is enabled
• otherwise, we enable force polling only if we detect we're running on WSL (Windows Subsystem for Linux)

Parameters:

Name	Type	Description	Default
*paths	Union[Path, str]	filesystem paths to watch.	()
watch_filter	Optional[Callable[[Change, str], bool]]	callable used to filter out changes which are not important, you can either use a raw callable or a BaseFilter instance, defaults to an instance of DefaultFilter. To keep all changes, use None.	DefaultFilter()
debounce	int	maximum time in milliseconds to group changes over before yielding them.	1600
step	int	time to wait for new changes in milliseconds, if no changes are detected in this time, and at least one change has been detected, the changes are yielded.	50
stop_event	Optional[AbstractEvent]	event to stop watching, if this is set, the generator will stop iteration, this can be anything with an is_set() method which returns a bool, e.g. threading.Event().	None
rust_timeout	int	maximum time in milliseconds to wait in the rust code for changes, 0 means no timeout.	5000
yield_on_timeout	bool	if True, the generator will yield upon timeout in rust even if no changes are detected.	False
debug	Optional[bool]	whether to print information about all filesystem changes in rust to stdout, if None will use the WATCHFILES_DEBUG environment variable.	None
raise_interrupt	bool	whether to re-raise KeyboardInterrupts, or suppress the error and just stop iterating.	True
force_polling	Optional[bool]	See Force polling above.	None
poll_delay_ms	int	delay between polling for changes, only used if force_polling=True.	300
recursive	bool	if True, watch for changes in sub-directories recursively, otherwise watch only for changes in the top-level directory, default is True.	True
ignore_permission_denied	Optional[bool]	if True, will ignore permission denied errors, otherwise will raise them by default. Setting the WATCHFILES_IGNORE_PERMISSION_DENIED environment variable will set this value too.	None

Yields:

Type	Description
Set[FileChange]	The generator yields sets of FileChanges.

Example of watch usage

from watchfiles import watch

for changes in watch('./first/dir', './second/dir', raise_interrupt=False):
    print(changes)

awatch async

awatch(*paths: Union[Path, str], watch_filter: Optional[Callable[[Change, str], bool]] = DefaultFilter(), debounce: int = 1600, step: int = 50, stop_event: Optional[AnyEvent] = None, rust_timeout: Optional[int] = None, yield_on_timeout: bool = False, debug: Optional[bool] = None, raise_interrupt: Optional[bool] = None, force_polling: Optional[bool] = None, poll_delay_ms: int = 300, recursive: bool = True, ignore_permission_denied: Optional[bool] = None) -> AsyncGenerator[Set[FileChange], None]

Asynchronous equivalent of watch using threads to wait for changes. Arguments match those of watch except stop_event.

All async methods use anyio to run the event loop.

Unlike watch KeyboardInterrupt cannot be suppressed by awatch so they need to be caught where asyncio.run or equivalent is called.

Parameters:

Name	Type	Description	Default
*paths	Union[Path, str]	filesystem paths to watch.	()
watch_filter	Optional[Callable[[Change, str], bool]]	matches the same argument of watch.	DefaultFilter()
debounce	int	matches the same argument of watch.	1600
step	int	matches the same argument of watch.	50
stop_event	Optional[AnyEvent]	anyio.Event which can be used to stop iteration, see example below.	None
rust_timeout	Optional[int]	matches the same argument of watch, except that None means use 1_000 on Windows and 5_000 on other platforms thus helping with exiting on Ctrl+C on Windows, see #110.	None
yield_on_timeout	bool	matches the same argument of watch.	False
debug	Optional[bool]	matches the same argument of watch.	None
raise_interrupt	Optional[bool]	This is deprecated, KeyboardInterrupt will cause this coroutine to be cancelled and then be raised by the top level asyncio.run call or equivalent, and should be caught there. See #136	None
force_polling	Optional[bool]	if true, always use polling instead of file system notifications, default is None where force_polling is set to True if the WATCHFILES_FORCE_POLLING environment variable exists.	None
poll_delay_ms	int	delay between polling for changes, only used if force_polling=True.	300
recursive	bool	if True, watch for changes in sub-directories recursively, otherwise watch only for changes in the top-level directory, default is True.	True
ignore_permission_denied	Optional[bool]	if True, will ignore permission denied errors, otherwise will raise them by default. Setting the WATCHFILES_IGNORE_PERMISSION_DENIED environment variable will set this value too.	None

Yields:

Type	Description
AsyncGenerator[Set[FileChange], None]	The generator yields sets of FileChanges.

Example of awatch usage

import asyncio
from watchfiles import awatch

async def main():
    async for changes in awatch('./first/dir', './second/dir'):
        print(changes)

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print('stopped via KeyboardInterrupt')

Example of awatch usage with a stop event

import asyncio
from watchfiles import awatch

async def main():
    stop_event = asyncio.Event()

    async def stop_soon():
        await asyncio.sleep(3)
        stop_event.set()

    stop_soon_task = asyncio.create_task(stop_soon())

    async for changes in awatch('/path/to/dir', stop_event=stop_event):
        print(changes)

    # cleanup by awaiting the (now complete) stop_soon_task
    await stop_soon_task

asyncio.run(main())

Change

Bases: IntEnum

Enum representing the type of change that occurred.

added class-attribute instance-attribute

added = 1

A new file or directory was added.

deleted class-attribute instance-attribute

deleted = 3

A file or directory was deleted.

modified class-attribute instance-attribute

modified = 2

A file or directory was modified, can be either a metadata or data change.

FileChange module-attribute

FileChange = Tuple[Change, str]

A tuple representing a file change, first element is a Change member, second is the path of the file or directory that changed.