codekingpro/portable-devtools

1from typing import Any, Literal, Protocol
2
3__all__ = 'RustNotify', 'WatchfilesRustInternalError'
4
5__version__: str
6"""The package version as defined in `Cargo.toml`, modified to match python's versioning semantics."""
7
8class AbstractEvent(Protocol):
9    def is_set(self) -> bool: ...
10
11class RustNotify:
12    """
13    Interface to the Rust [notify](https://crates.io/crates/notify) crate which does
14    the heavy lifting of watching for file changes and grouping them into events.
15    """
16
17    def __init__(
18        self,
19        watch_paths: list[str],
20        debug: bool,
21        force_polling: bool,
22        poll_delay_ms: int,
23        recursive: bool,
24        ignore_permission_denied: bool,
25    ) -> None:
26        """
27        Create a new `RustNotify` instance and start a thread to watch for changes.
28
29        `FileNotFoundError` is raised if any of the paths do not exist.
30
31        Args:
32            watch_paths: file system paths to watch for changes, can be directories or files
33            debug: if true, print details about all events to stderr
34            force_polling: if true, always use polling instead of file system notifications
35            poll_delay_ms: delay between polling for changes, only used if `force_polling=True`
36            recursive: if `True`, watch for changes in sub-directories recursively, otherwise watch only for changes in
37                the top-level directory, default is `True`.
38            ignore_permission_denied: if `True`, permission denied errors are ignored while watching changes.
39        """
40    def watch(
41        self,
42        debounce_ms: int,
43        step_ms: int,
44        timeout_ms: int,
45        stop_event: AbstractEvent | None,
46    ) -> set[tuple[int, str]] | Literal['signal', 'stop', 'timeout']:
47        """
48        Watch for changes.
49
50        This method will wait `timeout_ms` milliseconds for changes, but once a change is detected,
51        it will group changes and return in no more than `debounce_ms` milliseconds.
52
53        The GIL is released during a `step_ms` sleep on each iteration to avoid
54        blocking python.
55
56        Args:
57            debounce_ms: maximum time in milliseconds to group changes over before returning.
58            step_ms: time to wait for new changes in milliseconds, if no changes are detected
59                in this time, and at least one change has been detected, the changes are yielded.
60            timeout_ms: maximum time in milliseconds to wait for changes before returning,
61                `0` means wait indefinitely, `debounce_ms` takes precedence over `timeout_ms` once
62                a change is detected.
63            stop_event: event to check on every iteration to see if this function should return early.
64                The event should be an object which has an `is_set()` method which returns a boolean.
65
66        Returns:
67            See below.
68
69        Return values have the following meanings:
70
71        * Change details as a `set` of `(event_type, path)` tuples, the event types are ints which match
72          [`Change`][watchfiles.Change], `path` is a string representing the path of the file that changed
73        * `'signal'` string, if a signal was received
74        * `'stop'` string, if the `stop_event` was set
75        * `'timeout'` string, if `timeout_ms` was exceeded
76        """
77    def __enter__(self) -> RustNotify:
78        """
79        Does nothing, but allows `RustNotify` to be used as a context manager.
80
81        !!! note
82
83            The watching thead is created when an instance is initiated, not on `__enter__`.
84        """
85    def __exit__(self, *args: Any) -> None:
86        """
87        Calls [`close`][watchfiles._rust_notify.RustNotify.close].
88        """
89    def close(self) -> None:
90        """
91        Stops the watching thread. After `close` is called, the `RustNotify` instance can no
92        longer be used, calls to [`watch`][watchfiles._rust_notify.RustNotify.watch] will raise a `RuntimeError`.
93
94        !!! note
95
96            `close` is not required, just deleting the `RustNotify` instance will kill the thread
97            implicitly.
98
99            As per [#163](https://github.com/samuelcolvin/watchfiles/issues/163) `close()` is only required because
100            in the event of an error, the traceback in `sys.exc_info` keeps a reference to `watchfiles.watch`'s
101            frame, so you can't rely on the `RustNotify` object being deleted, and thereby stopping
102            the watching thread.
103        """
104
105class WatchfilesRustInternalError(RuntimeError):
106    """
107    Raised when RustNotify encounters an unknown error.
108
109    If you get this a lot, please check [github](https://github.com/samuelcolvin/watchfiles/issues) issues
110    and create a new issue if your problem is not discussed.
111    """
112