Skip to content

RustNotify

RustNotify(
    watch_paths: List[str],
    debug: bool,
    force_polling: bool,
    poll_delay_ms: int,
    recursive: 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

__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)

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) 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.)