RustNotify ¶

RustNotify(
    watch_paths: List[str],
    debug: bool,
    force_polling: bool,
    poll_delay_ms: int,
    recursive: bool,
    ignore_permission_denied: bool,
) -> None

Interface to the Rust notify crate which does the heavy lifting of watching for file changes and grouping them into events.

Create a new RustNotify instance and start a thread to watch for changes.

FileNotFoundError is raised if any of the paths do not exist.

Parameters:

Name	Type	Description	Default
`watch_paths`	`List[str]`	file system paths to watch for changes, can be directories or files	required
`debug`	`bool`	if true, print details about all events to stderr	required
`force_polling`	`bool`	if true, always use polling instead of file system notifications	required
`poll_delay_ms`	`int`	delay between polling for changes, only used if `force_polling=True`	required
`recursive`	`bool`	if `True`, watch for changes in sub-directories recursively, otherwise watch only for changes in the top-level directory, default is `True`.	required
`ignore_permission_denied`	`bool`	if `True`, permission denied errors are ignored while watching changes.	required

enter ¶

__enter__() -> RustNotify

Does nothing, but allows RustNotify to be used as a context manager.

Note

The watching thead is created when an instance is initiated, not on __enter__.

exit ¶

__exit__(*args: Any) -> None

Calls close.

close ¶

close() -> None

Stops the watching thread. After close is called, the RustNotify instance can no longer be used, calls to watch will raise a RuntimeError.

Note

close is not required, just deleting the RustNotify instance will kill the thread implicitly.

As per #163 close() is only required because in the event of an error, the traceback in sys.exc_info keeps a reference to watchfiles.watch's frame, so you can't rely on the RustNotify object being deleted, and thereby stopping the watching thread.

watch ¶

watch(
    debounce_ms: int,
    step_ms: int,
    timeout_ms: int,
    stop_event: Optional[AbstractEvent],
) -> Union[
    Set[Tuple[int, str]],
    Literal["signal", "stop", "timeout"],
]

Watch for changes.

This method will wait timeout_ms milliseconds for changes, but once a change is detected, it will group changes and return in no more than debounce_ms milliseconds.

The GIL is released during a step_ms sleep on each iteration to avoid blocking python.

Parameters:

Name	Type	Description	Default
`debounce_ms`	`int`	maximum time in milliseconds to group changes over before returning.	required
`step_ms`	`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.	required
`timeout_ms`	`int`	maximum time in milliseconds to wait for changes before returning, `0` means wait indefinitely, `debounce_ms` takes precedence over `timeout_ms` once a change is detected.	required
`stop_event`	`Optional[AbstractEvent]`	event to check on every iteration to see if this function should return early. The event should be an object which has an `is_set()` method which returns a boolean.	required

Returns:

Type	Description
`Union[Set[Tuple[int, str]], Literal['signal', 'stop', 'timeout']]`	See below.

Return values have the following meanings:

Change details as a set of (event_type, path) tuples, the event types are ints which match Change, path is a string representing the path of the file that changed
'signal' string, if a signal was received
'stop' string, if the stop_event was set
'timeout' string, if timeout_ms was exceeded

WatchfilesRustInternalError ¶

Bases: RuntimeError

Raised when RustNotify encounters an unknown error.

If you get this a lot, please check github issues and create a new issue if your problem is not discussed.

version `module-attribute` ¶

__version__: str

The package version as defined in Cargo.toml, modified to match python's versioning semantics.

Rust backend direct usage¶

The rust backend can be accessed directly as follows:

Rust backend example

from watchfiles._rust_notify import RustNotify

r = RustNotify(['first/path', 'second/path'], False, False, 0, True, False)

changes = r.watch(1_600, 50, 100, None)
print(changes)

r.close()

Or using RustNotify as a context manager:

Rust backend context manager example

from watchfiles._rust_notify import RustNotify

with RustNotify(['first/path', 'second/path'], False, False, 0, True, False) as r:
    changes = r.watch(1_600, 50, 100, None)
    print(changes)

(See the documentation on close above for when and why the context manager or close method are required.)