from __future__ import annotations from pathlib import Path from socket import socket from typing import Callable from watchfiles import watch from uvicorn.config import Config from uvicorn.supervisors.basereload import BaseReload class FileFilter: def __init__(self, config: Config): default_includes = ["*.py"] self.includes = [ default for default in default_includes if default not in config.reload_excludes ] self.includes.extend(config.reload_includes) self.includes = list(set(self.includes)) default_excludes = [".*", ".py[cod]", ".sw.*", "~*"] self.excludes = [ default for default in default_excludes if default not in config.reload_includes ] self.exclude_dirs = [] for e in config.reload_excludes: p = Path(e) try: is_dir = p.is_dir() except OSError: # pragma: no cover # gets raised on Windows for values like "*.py" is_dir = False if is_dir: self.exclude_dirs.append(p) else: self.excludes.append(e) self.excludes = list(set(self.excludes)) def __call__(self, path: Path) -> bool: for include_pattern in self.includes: if path.match(include_pattern): if str(path).endswith(include_pattern): return True for exclude_dir in self.exclude_dirs: if exclude_dir in path.parents: return False for exclude_pattern in self.excludes: if path.match(exclude_pattern): return False return True return False class WatchFilesReload(BaseReload): def __init__( self, config: Config, target: Callable[[list[socket] | None], None], sockets: list[socket], ) -> None: super().__init__(config, target, sockets) self.reloader_name = "WatchFiles" self.reload_dirs = [] for directory in config.reload_dirs: if Path.cwd() not in directory.parents: self.reload_dirs.append(directory) if Path.cwd() not in self.reload_dirs: self.reload_dirs.append(Path.cwd()) self.watch_filter = FileFilter(config) self.watcher = watch( *self.reload_dirs, watch_filter=None, stop_event=self.should_exit, # using yield_on_timeout here mostly to make sure tests don't # hang forever, won't affect the class's behavior yield_on_timeout=True, ) def should_restart(self) -> list[Path] | None: self.pause() changes = next(self.watcher) if changes: unique_paths = {Path(c[1]) for c in changes} return [p for p in unique_paths if self.watch_filter(p)] return None