codekingpro/portable-devtools

1from __future__ import annotations2 3import fnmatch4import os5import subprocess6import sys7import threading8import time9import typing as t10from itertools import chain11from pathlib import PurePath12 13from ._internal import _log14 15# The various system prefixes where imports are found. Base values are16# different when running in a virtualenv. All reloaders will ignore the17# base paths (usually the system installation). The stat reloader won't18# scan the virtualenv paths, it will only include modules that are19# already imported.20_ignore_always = tuple({sys.base_prefix, sys.base_exec_prefix})21prefix = {*_ignore_always, sys.prefix, sys.exec_prefix}22 23if hasattr(sys, "real_prefix"):24    # virtualenv < 2025    prefix.add(sys.real_prefix)26 27_stat_ignore_scan = tuple(prefix)28del prefix29# Ignore __pycache__ since a change there will always have a change to30# the source file (or initial pyc file) as well. Ignore common version control31# internals. Ignore common tool caches.32_ignore_common_dirs = {33    "__pycache__",34    ".git",35    ".hg",36    ".tox",37    ".nox",38    ".pytest_cache",39    ".mypy_cache",40}41 42 43def _iter_module_paths() -> t.Iterator[str]:44    """Find the filesystem paths associated with imported modules."""45    # List is in case the value is modified by the app while updating.46    for module in list(sys.modules.values()):47        name = getattr(module, "__file__", None)48 49        if name is None or name.startswith(_ignore_always):50            continue51 52        while not os.path.isfile(name):53            # Zip file, find the base file without the module path.54            old = name55            name = os.path.dirname(name)56 57            if name == old:  # skip if it was all directories somehow58                break59        else:60            yield name61 62 63def _remove_by_pattern(paths: set[str], exclude_patterns: set[str]) -> None:64    for pattern in exclude_patterns:65        paths.difference_update(fnmatch.filter(paths, pattern))66 67 68def _find_stat_paths(69    extra_files: set[str], exclude_patterns: set[str]70) -> t.Iterable[str]:71    """Find paths for the stat reloader to watch. Returns imported72    module files, Python files under non-system paths. Extra files and73    Python files under extra directories can also be scanned.74 75    System paths have to be excluded for efficiency. Non-system paths,76    such as a project root or ``sys.path.insert``, should be the paths77    of interest to the user anyway.78    """79    paths = set()80 81    for path in chain(list(sys.path), extra_files):82        path = os.path.abspath(path)83 84        if os.path.isfile(path):85            # zip file on sys.path, or extra file86            paths.add(path)87            continue88 89        parent_has_py = {os.path.dirname(path): True}90 91        for root, dirs, files in os.walk(path):92            if (93                root.startswith(_stat_ignore_scan)94                or os.path.basename(root) in _ignore_common_dirs95            ):96                dirs.clear()97                continue98 99            has_py = False100 101            for name in files:102                if name.endswith((".py", ".pyc")):103                    has_py = True104                    paths.add(os.path.join(root, name))105 106            # Optimization: stop scanning a directory if neither it nor107            # its parent contained Python files.108            if not (has_py or parent_has_py[os.path.dirname(root)]):109                dirs.clear()110                continue111 112            parent_has_py[root] = has_py113 114    paths.update(_iter_module_paths())115    _remove_by_pattern(paths, exclude_patterns)116    return paths117 118 119def _find_watchdog_paths(120    extra_files: set[str], exclude_patterns: set[str]121) -> t.Iterable[str]:122    """Find paths for the stat reloader to watch. Looks at the same123    sources as the stat reloader, but watches everything under124    directories instead of individual files.125    """126    dirs = set()127 128    for name in chain(list(sys.path), extra_files):129        name = os.path.abspath(name)130 131        if os.path.isfile(name):132            name = os.path.dirname(name)133 134        dirs.add(name)135 136    for name in _iter_module_paths():137        dirs.add(os.path.dirname(name))138 139    _remove_by_pattern(dirs, exclude_patterns)140    return _find_common_roots(dirs)141 142 143def _find_common_roots(paths: t.Iterable[str]) -> t.Iterable[str]:144    root: dict[str, dict[str, t.Any]] = {}145 146    for chunks in sorted((PurePath(x).parts for x in paths), key=len, reverse=True):147        node = root148 149        for chunk in chunks:150            node = node.setdefault(chunk, {})151 152        node.clear()153 154    rv = set()155 156    def _walk(node: t.Mapping[str, dict[str, t.Any]], path: tuple[str, ...]) -> None:157        for prefix, child in node.items():158            _walk(child, path + (prefix,))159 160        # If there are no more nodes, and a path has been accumulated, add it.161        # Path may be empty if the "" entry is in sys.path.162        if not node and path:163            rv.add(os.path.join(*path))164 165    _walk(root, ())166    return rv167 168 169def _get_args_for_reloading() -> list[str]:170    """Determine how the script was executed, and return the args needed171    to execute it again in a new process.172    """173    if sys.version_info >= (3, 10):174        # sys.orig_argv, added in Python 3.10, contains the exact args used to invoke175        # Python. Still replace argv[0] with sys.executable for accuracy.176        return [sys.executable, *sys.orig_argv[1:]]177 178    rv = [sys.executable]179    py_script = sys.argv[0]180    args = sys.argv[1:]181    # Need to look at main module to determine how it was executed.182    __main__ = sys.modules["__main__"]183 184    # The value of __package__ indicates how Python was called. It may185    # not exist if a setuptools script is installed as an egg. It may be186    # set incorrectly for entry points created with pip on Windows.187    if getattr(__main__, "__package__", None) is None or (188        os.name == "nt"189        and __main__.__package__ == ""190        and not os.path.exists(py_script)191        and os.path.exists(f"{py_script}.exe")192    ):193        # Executed a file, like "python app.py".194        py_script = os.path.abspath(py_script)195 196        if os.name == "nt":197            # Windows entry points have ".exe" extension and should be198            # called directly.199            if not os.path.exists(py_script) and os.path.exists(f"{py_script}.exe"):200                py_script += ".exe"201 202            if (203                os.path.splitext(sys.executable)[1] == ".exe"204                and os.path.splitext(py_script)[1] == ".exe"205            ):206                rv.pop(0)207 208        rv.append(py_script)209    else:210        # Executed a module, like "python -m werkzeug.serving".211        if os.path.isfile(py_script):212            # Rewritten by Python from "-m script" to "/path/to/script.py".213            py_module = t.cast(str, __main__.__package__)214            name = os.path.splitext(os.path.basename(py_script))[0]215 216            if name != "__main__":217                py_module += f".{name}"218        else:219            # Incorrectly rewritten by pydevd debugger from "-m script" to "script".220            py_module = py_script221 222        rv.extend(("-m", py_module.lstrip(".")))223 224    rv.extend(args)225    return rv226 227 228class ReloaderLoop:229    name = ""230 231    def __init__(232        self,233        extra_files: t.Iterable[str] | None = None,234        exclude_patterns: t.Iterable[str] | None = None,235        interval: int | float = 1,236    ) -> None:237        self.extra_files: set[str] = {os.path.abspath(x) for x in extra_files or ()}238        self.exclude_patterns: set[str] = set(exclude_patterns or ())239        self.interval = interval240 241    def __enter__(self) -> ReloaderLoop:242        """Do any setup, then run one step of the watch to populate the243        initial filesystem state.244        """245        self.run_step()246        return self247 248    def __exit__(self, exc_type, exc_val, exc_tb):  # type: ignore249        """Clean up any resources associated with the reloader."""250        pass251 252    def run(self) -> None:253        """Continually run the watch step, sleeping for the configured254        interval after each step.255        """256        while True:257            self.run_step()258            time.sleep(self.interval)259 260    def run_step(self) -> None:261        """Run one step for watching the filesystem. Called once to set262        up initial state, then repeatedly to update it.263        """264        pass265 266    def restart_with_reloader(self) -> int:267        """Spawn a new Python interpreter with the same arguments as the268        current one, but running the reloader thread.269        """270        while True:271            _log("info", f" * Restarting with {self.name}")272            args = _get_args_for_reloading()273            new_environ = os.environ.copy()274            new_environ["WERKZEUG_RUN_MAIN"] = "true"275            exit_code = subprocess.call(args, env=new_environ, close_fds=False)276 277            if exit_code != 3:278                return exit_code279 280    def trigger_reload(self, filename: str) -> None:281        self.log_reload(filename)282        sys.exit(3)283 284    def log_reload(self, filename: str | bytes) -> None:285        filename = os.path.abspath(filename)286        _log("info", f" * Detected change in {filename!r}, reloading")287 288 289class StatReloaderLoop(ReloaderLoop):290    name = "stat"291 292    def __enter__(self) -> ReloaderLoop:293        self.mtimes: dict[str, float] = {}294        return super().__enter__()295 296    def run_step(self) -> None:297        for name in _find_stat_paths(self.extra_files, self.exclude_patterns):298            try:299                mtime = os.stat(name).st_mtime300            except OSError:301                continue302 303            old_time = self.mtimes.get(name)304 305            if old_time is None:306                self.mtimes[name] = mtime307                continue308 309            if mtime > old_time:310                self.trigger_reload(name)311 312 313class WatchdogReloaderLoop(ReloaderLoop):314    def __init__(self, *args: t.Any, **kwargs: t.Any) -> None:315        from watchdog.events import EVENT_TYPE_CLOSED316        from watchdog.events import EVENT_TYPE_CREATED317        from watchdog.events import EVENT_TYPE_DELETED318        from watchdog.events import EVENT_TYPE_MODIFIED319        from watchdog.events import EVENT_TYPE_MOVED320        from watchdog.events import FileModifiedEvent321        from watchdog.events import PatternMatchingEventHandler322        from watchdog.observers import Observer323 324        super().__init__(*args, **kwargs)325        trigger_reload = self.trigger_reload326 327        class EventHandler(PatternMatchingEventHandler):328            def on_any_event(self, event: FileModifiedEvent) -> None:  # type: ignore[override]329                if event.event_type not in {330                    EVENT_TYPE_CLOSED,331                    EVENT_TYPE_CREATED,332                    EVENT_TYPE_DELETED,333                    EVENT_TYPE_MODIFIED,334                    EVENT_TYPE_MOVED,335                }:336                    # skip events that don't involve changes to the file337                    return338 339                trigger_reload(event.src_path)340 341        reloader_name = Observer.__name__.lower()  # type: ignore[attr-defined]342 343        if reloader_name.endswith("observer"):344            reloader_name = reloader_name[:-8]345 346        self.name = f"watchdog ({reloader_name})"347        self.observer = Observer()348        extra_patterns = (p for p in self.extra_files if not os.path.isdir(p))349        self.event_handler = EventHandler(350            patterns=["*.py", "*.pyc", "*.zip", *extra_patterns],351            ignore_patterns=[352                *[f"*/{d}/*" for d in _ignore_common_dirs],353                *self.exclude_patterns,354            ],355        )356        self.should_reload = threading.Event()357 358    def trigger_reload(self, filename: str | bytes) -> None:359        # This is called inside an event handler, which means throwing360        # SystemExit has no effect.361        # https://github.com/gorakhargosh/watchdog/issues/294362        self.should_reload.set()363        self.log_reload(filename)364 365    def __enter__(self) -> ReloaderLoop:366        self.watches: dict[str, t.Any] = {}367        self.observer.start()368        return super().__enter__()369 370    def __exit__(self, exc_type, exc_val, exc_tb):  # type: ignore371        self.observer.stop()372        self.observer.join()373 374    def run(self) -> None:375        while not self.should_reload.wait(timeout=self.interval):376            self.run_step()377 378        sys.exit(3)379 380    def run_step(self) -> None:381        to_delete = set(self.watches)382 383        for path in _find_watchdog_paths(self.extra_files, self.exclude_patterns):384            if path not in self.watches:385                try:386                    self.watches[path] = self.observer.schedule(387                        self.event_handler, path, recursive=True388                    )389                except OSError:390                    # Clear this path from list of watches. We don't want391                    # the same error message showing again in the next392                    # iteration.393                    self.watches[path] = None394 395            to_delete.discard(path)396 397        for path in to_delete:398            watch = self.watches.pop(path, None)399 400            if watch is not None:401                self.observer.unschedule(watch)402 403 404reloader_loops: dict[str, type[ReloaderLoop]] = {405    "stat": StatReloaderLoop,406    "watchdog": WatchdogReloaderLoop,407}408 409try:410    __import__("watchdog.observers")411except ImportError:412    reloader_loops["auto"] = reloader_loops["stat"]413else:414    reloader_loops["auto"] = reloader_loops["watchdog"]415 416 417def ensure_echo_on() -> None:418    """Ensure that echo mode is enabled. Some tools such as PDB disable419    it which causes usability issues after a reload."""420    # tcgetattr will fail if stdin isn't a tty421    if sys.stdin is None or not sys.stdin.isatty():422        return423 424    try:425        import termios426    except ImportError:427        return428 429    attributes = termios.tcgetattr(sys.stdin)430 431    if not attributes[3] & termios.ECHO:432        attributes[3] |= termios.ECHO433        termios.tcsetattr(sys.stdin, termios.TCSANOW, attributes)434 435 436def run_with_reloader(437    main_func: t.Callable[[], None],438    extra_files: t.Iterable[str] | None = None,439    exclude_patterns: t.Iterable[str] | None = None,440    interval: int | float = 1,441    reloader_type: str = "auto",442) -> None:443    """Run the given function in an independent Python interpreter."""444    import signal445 446    signal.signal(signal.SIGTERM, lambda *args: sys.exit(0))447    reloader = reloader_loops[reloader_type](448        extra_files=extra_files, exclude_patterns=exclude_patterns, interval=interval449    )450 451    try:452        if os.environ.get("WERKZEUG_RUN_MAIN") == "true":453            ensure_echo_on()454            t = threading.Thread(target=main_func, args=())455            t.daemon = True456 457            # Enter the reloader to set up initial state, then start458            # the app thread and reloader update loop.459            with reloader:460                t.start()461                reloader.run()462        else:463            sys.exit(reloader.restart_with_reloader())464    except KeyboardInterrupt:465        pass466