Skip to content

WatchReloader

faststream.cli.supervisors.watchfiles.WatchReloader #

WatchReloader(
    target: DecoratedCallable,
    args: Tuple[Any, ...],
    reload_dirs: Sequence[Union[Path, str]],
    reload_delay: float = 0.3,
)

Bases: BaseReload

A class to reload a target function when files in specified directories change.

METHOD DESCRIPTION
should_restart

Checks if any files in the watched directories have changed and returns True if a change is detected, False otherwise.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Initialize a WatchFilesReloader object.

PARAMETER DESCRIPTION
target

The target callable to be executed.

TYPE: DecoratedCallable

args

The arguments to be passed to the target callable.

TYPE: Tuple[Any, ...]

reload_dirs

A sequence of directories to watch for changes.

TYPE: Sequence[Union[Path, str]]

reload_delay

The delay in seconds between checking for changes. Default is 0.3.

TYPE: float DEFAULT: 0.3

RETURNS DESCRIPTION
None

None.

Note

The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)

Source code in faststream/cli/supervisors/watchfiles.py 
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
def __init__(
    self,
    target: DecoratedCallable,
    args: Tuple[Any, ...],
    reload_dirs: Sequence[Union[Path, str]],
    reload_delay: float = 0.3,
) -> None:
    """Initialize a WatchFilesReloader object.

    Args:
        target: The target callable to be executed.
        args: The arguments to be passed to the target callable.
        reload_dirs: A sequence of directories to watch for changes.
        reload_delay: The delay in seconds between checking for changes. Default is 0.3.

    Returns:
        None.
    !!! note

        The above docstring is autogenerated by docstring-gen library (https://docstring-gen.airt.ai)
    """
    super().__init__(target, args, reload_delay)
    self.reloader_name = "WatchFiles"
    self.reload_dirs = reload_dirs
    self.watcher = watchfiles.watch(
        *reload_dirs,
        step=int(reload_delay * 1000),
        watch_filter=ExtendedFilter(),
        stop_event=self.should_exit,
        yield_on_timeout=True,
    )

pid instance-attribute #

pid: int = os.getpid()

reload_delay instance-attribute #

reload_delay: Optional[float] = reload_delay

reload_dirs instance-attribute #

reload_dirs = reload_dirs

reloader_name instance-attribute #

reloader_name = 'WatchFiles'

should_exit instance-attribute #

should_exit: threading.Event = threading.Event()

watcher instance-attribute #

watcher = watchfiles.watch(
    *reload_dirs,
    step=int(reload_delay * 1000),
    watch_filter=ExtendedFilter(),
    stop_event=self.should_exit,
    yield_on_timeout=True
)

restart #

restart() -> None
Source code in faststream/cli/supervisors/basereload.py 
100
101
102
103
def restart(self) -> None:
    self._stop_process()
    logger.info("Process successfully reloaded")
    self._process = self._start_process()

run #

run() -> None
Source code in faststream/cli/supervisors/basereload.py 
89
90
91
92
93
94
def run(self) -> None:
    self.startup()
    while not self.should_exit.wait(self.reload_delay):
        if self.should_restart():  # pragma: no branch
            self.restart()
    self.shutdown()

should_restart #

should_restart() -> bool
Source code in faststream/cli/supervisors/watchfiles.py 
114
115
116
117
118
119
120
121
def should_restart(self) -> bool:
    for changes in self.watcher:  # pragma: no branch
        if changes:  # pragma: no branch
            unique_paths = {Path(c[1]).name for c in changes}
            message = "WatchReloader detected file change in '%s'. Reloading..."
            logger.info(message % tuple(unique_paths))
            return True
    return False  # pragma: no cover

shutdown #

shutdown() -> None
Source code in faststream/cli/supervisors/basereload.py 
105
106
107
def shutdown(self) -> None:
    self._stop_process()
    logger.info(f"Stopping reloader process [{self.pid}]")

startup #

startup() -> None
Source code in faststream/cli/supervisors/watchfiles.py 
110
111
112
def startup(self) -> None:
    logger.info(f"Will watch for changes in these directories: {self.reload_dirs}")
    super().startup()
Last update: 2023-11-13