codekingpro/portable-devtools

1from __future__ import annotations2 3from collections.abc import Callable4from pathlib import Path5from socket import socket6 7from watchfiles import watch8 9from uvicorn.config import Config10from uvicorn.supervisors.basereload import BaseReload11 12 13class FileFilter:14    def __init__(self, config: Config):15        default_includes = ["*.py"]16        self.includes = [default for default in default_includes if default not in config.reload_excludes]17        self.includes.extend(config.reload_includes)18        self.includes = list(set(self.includes))19 20        default_excludes = [".*", ".py[cod]", ".sw.*", "~*"]21        self.excludes = [default for default in default_excludes if default not in config.reload_includes]22        self.exclude_dirs = []23        for e in config.reload_excludes:24            p = Path(e)25            try:26                is_dir = p.is_dir()27            except OSError:  # pragma: no cover28                # gets raised on Windows for values like "*.py"29                is_dir = False30 31            if is_dir:32                self.exclude_dirs.append(p)33            else:34                self.excludes.append(e)  # pragma: full coverage35        self.excludes = list(set(self.excludes))36 37    def __call__(self, path: Path) -> bool:38        for include_pattern in self.includes:39            if path.match(include_pattern):40                if str(path).endswith(include_pattern):41                    return True  # pragma: full coverage42 43                for exclude_dir in self.exclude_dirs:44                    if exclude_dir in path.parents:45                        return False46 47                for exclude_pattern in self.excludes:48                    if path.match(exclude_pattern):49                        return False  # pragma: full coverage50 51                return True52        return False53 54 55class WatchFilesReload(BaseReload):56    def __init__(57        self,58        config: Config,59        target: Callable[[list[socket] | None], None],60        sockets: list[socket],61    ) -> None:62        super().__init__(config, target, sockets)63        self.reloader_name = "WatchFiles"64        self.reload_dirs: list[Path] = []65        for directory in config.reload_dirs:66            self.reload_dirs.append(directory)67 68        self.watch_filter = FileFilter(config)69        self.watcher = watch(70            *self.reload_dirs,71            watch_filter=None,72            stop_event=self.should_exit,73            # using yield_on_timeout here mostly to make sure tests don't74            # hang forever, won't affect the class's behavior75            yield_on_timeout=True,76            ignore_permission_denied=True,77        )78 79    def should_restart(self) -> list[Path] | None:80        self.pause()81 82        changes = next(self.watcher)83        if changes:84            unique_paths = {Path(c[1]) for c in changes}85            return [p for p in unique_paths if self.watch_filter(p)]86        return None87