JezK
Edit File: _reloader.py
import fnmatch import os import subprocess import sys import threading import time import typing as t from itertools import chain from pathlib import PurePath from ._internal import _log # The various system prefixes where imports are found. Base values are # different when running in a virtualenv. All reloaders will ignore the # base paths (usually the system installation). The stat reloader won't # scan the virtualenv paths, it will only include modules that are # already imported. _ignore_always = tuple({sys.base_prefix, sys.base_exec_prefix}) prefix = {*_ignore_always, sys.prefix, sys.exec_prefix} if hasattr(sys, "real_prefix"): # virtualenv < 20 prefix.add(sys.real_prefix) # type: ignore[attr-defined] _stat_ignore_scan = tuple(prefix) del prefix _ignore_common_dirs = { "__pycache__", ".git", ".hg", ".tox", ".nox", ".pytest_cache", ".mypy_cache", } def _iter_module_paths() -> t.Iterator[str]: """Find the filesystem paths associated with imported modules.""" # List is in case the value is modified by the app while updating. for module in list(sys.modules.values()): name = getattr(module, "__file__", None) if name is None or name.startswith(_ignore_always): continue while not os.path.isfile(name): # Zip file, find the base file without the module path. old = name name = os.path.dirname(name) if name == old: # skip if it was all directories somehow break else: yield name def _remove_by_pattern(paths: t.Set[str], exclude_patterns: t.Set[str]) -> None: for pattern in exclude_patterns: paths.difference_update(fnmatch.filter(paths, pattern)) def _find_stat_paths( extra_files: t.Set[str], exclude_patterns: t.Set[str] ) -> t.Iterable[str]: """Find paths for the stat reloader to watch. Returns imported module files, Python files under non-system paths. Extra files and Python files under extra directories can also be scanned. System paths have to be excluded for efficiency. Non-system paths, such as a project root or ``sys.path.insert``, should be the paths of interest to the user anyway. """ paths = set() for path in chain(list(sys.path), extra_files): path = os.path.abspath(path) if os.path.isfile(path): # zip file on sys.path, or extra file paths.add(path) continue parent_has_py = {os.path.dirname(path): True} for root, dirs, files in os.walk(path): # Optimizations: ignore system prefixes, __pycache__ will # have a py or pyc module at the import path, ignore some # common known dirs such as version control and tool caches. if ( root.startswith(_stat_ignore_scan) or os.path.basename(root) in _ignore_common_dirs ): dirs.clear() continue has_py = False for name in files: if name.endswith((".py", ".pyc")): has_py = True paths.add(os.path.join(root, name)) # Optimization: stop scanning a directory if neither it nor # its parent contained Python files. if not (has_py or parent_has_py[os.path.dirname(root)]): dirs.clear() continue parent_has_py[root] = has_py paths.update(_iter_module_paths()) _remove_by_pattern(paths, exclude_patterns) return paths def _find_watchdog_paths( extra_files: t.Set[str], exclude_patterns: t.Set[str] ) -> t.Iterable[str]: """Find paths for the stat reloader to watch. Looks at the same sources as the stat reloader, but watches everything under directories instead of individual files. """ dirs = set() for name in chain(list(sys.path), extra_files): name = os.path.abspath(name) if os.path.isfile(name): name = os.path.dirname(name) dirs.add(name) for name in _iter_module_paths(): dirs.add(os.path.dirname(name)) _remove_by_pattern(dirs, exclude_patterns) return _find_common_roots(dirs) def _find_common_roots(paths: t.Iterable[str]) -> t.Iterable[str]: root: t.Dict[str, dict] = {} for chunks in sorted((PurePath(x).parts for x in paths), key=len, reverse=True): node = root for chunk in chunks: node = node.setdefault(chunk, {}) node.clear() rv = set() def _walk(node: t.Mapping[str, dict], path: t.Tuple[str, ...]) -> None: for prefix, child in node.items(): _walk(child, path + (prefix,)) if not node: rv.add(os.path.join(*path)) _walk(root, ()) return rv def _get_args_for_reloading() -> t.List[str]: """Determine how the script was executed, and return the args needed to execute it again in a new process. """ rv = [sys.executable] py_script = sys.argv[0] args = sys.argv[1:] # Need to look at main module to determine how it was executed. __main__ = sys.modules["__main__"] # The value of __package__ indicates how Python was called. It may # not exist if a setuptools script is installed as an egg. It may be # set incorrectly for entry points created with pip on Windows. if getattr(__main__, "__package__", None) is None or ( os.name == "nt" and __main__.__package__ == "" and not os.path.exists(py_script) and os.path.exists(f"{py_script}.exe") ): # Executed a file, like "python app.py". py_script = os.path.abspath(py_script) if os.name == "nt": # Windows entry points have ".exe" extension and should be # called directly. if not os.path.exists(py_script) and os.path.exists(f"{py_script}.exe"): py_script += ".exe" if ( os.path.splitext(sys.executable)[1] == ".exe" and os.path.splitext(py_script)[1] == ".exe" ): rv.pop(0) rv.append(py_script) else: # Executed a module, like "python -m werkzeug.serving". if os.path.isfile(py_script): # Rewritten by Python from "-m script" to "/path/to/script.py". py_module = t.cast(str, __main__.__package__) name = os.path.splitext(os.path.basename(py_script))[0] if name != "__main__": py_module += f".{name}" else: # Incorrectly rewritten by pydevd debugger from "-m script" to "script". py_module = py_script rv.extend(("-m", py_module.lstrip("."))) rv.extend(args) return rv class ReloaderLoop: name = "" def __init__( self, extra_files: t.Optional[t.Iterable[str]] = None, exclude_patterns: t.Optional[t.Iterable[str]] = None, interval: t.Union[int, float] = 1, ) -> None: self.extra_files: t.Set[str] = {os.path.abspath(x) for x in extra_files or ()} self.exclude_patterns: t.Set[str] = set(exclude_patterns or ()) self.interval = interval def __enter__(self) -> "ReloaderLoop": """Do any setup, then run one step of the watch to populate the initial filesystem state. """ self.run_step() return self def __exit__(self, exc_type, exc_val, exc_tb): # type: ignore """Clean up any resources associated with the reloader.""" pass def run(self) -> None: """Continually run the watch step, sleeping for the configured interval after each step. """ while True: self.run_step() time.sleep(self.interval) def run_step(self) -> None: """Run one step for watching the filesystem. Called once to set up initial state, then repeatedly to update it. """ pass def restart_with_reloader(self) -> int: """Spawn a new Python interpreter with the same arguments as the current one, but running the reloader thread. """ while True: _log("info", f" * Restarting with {self.name}") args = _get_args_for_reloading() new_environ = os.environ.copy() new_environ["WERKZEUG_RUN_MAIN"] = "true" exit_code = subprocess.call(args, env=new_environ, close_fds=False) if exit_code != 3: return exit_code def trigger_reload(self, filename: str) -> None: self.log_reload(filename) sys.exit(3) def log_reload(self, filename: str) -> None: filename = os.path.abspath(filename) _log("info", f" * Detected change in {filename!r}, reloading") class StatReloaderLoop(ReloaderLoop): name = "stat" def __enter__(self) -> ReloaderLoop: self.mtimes: t.Dict[str, float] = {} return super().__enter__() def run_step(self) -> None: for name in _find_stat_paths(self.extra_files, self.exclude_patterns): try: mtime = os.stat(name).st_mtime except OSError: continue old_time = self.mtimes.get(name) if old_time is None: self.mtimes[name] = mtime continue if mtime > old_time: self.trigger_reload(name) class WatchdogReloaderLoop(ReloaderLoop): def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: from watchdog.observers import Observer from watchdog.events import PatternMatchingEventHandler super().__init__(*args, **kwargs) trigger_reload = self.trigger_reload class EventHandler(PatternMatchingEventHandler): # type: ignore def on_any_event(self, event): # type: ignore trigger_reload(event.src_path) reloader_name = Observer.__name__.lower() if reloader_name.endswith("observer"): reloader_name = reloader_name[:-8] self.name = f"watchdog ({reloader_name})" self.observer = Observer() # Extra patterns can be non-Python files, match them in addition # to all Python files in default and extra directories. Ignore # __pycache__ since a change there will always have a change to # the source file (or initial pyc file) as well. Ignore Git and # Mercurial internal changes. extra_patterns = [p for p in self.extra_files if not os.path.isdir(p)] self.event_handler = EventHandler( patterns=["*.py", "*.pyc", "*.zip", *extra_patterns], ignore_patterns=[ *[f"*/{d}/*" for d in _ignore_common_dirs], *self.exclude_patterns, ], ) self.should_reload = False def trigger_reload(self, filename: str) -> None: # This is called inside an event handler, which means throwing # SystemExit has no effect. # https://github.com/gorakhargosh/watchdog/issues/294 self.should_reload = True self.log_reload(filename) def __enter__(self) -> ReloaderLoop: self.watches: t.Dict[str, t.Any] = {} self.observer.start() return super().__enter__() def __exit__(self, exc_type, exc_val, exc_tb): # type: ignore self.observer.stop() self.observer.join() def run(self) -> None: while not self.should_reload: self.run_step() time.sleep(self.interval) sys.exit(3) def run_step(self) -> None: to_delete = set(self.watches) for path in _find_watchdog_paths(self.extra_files, self.exclude_patterns): if path not in self.watches: try: self.watches[path] = self.observer.schedule( self.event_handler, path, recursive=True ) except OSError: # Clear this path from list of watches We don't want # the same error message showing again in the next # iteration. self.watches[path] = None to_delete.discard(path) for path in to_delete: watch = self.watches.pop(path, None) if watch is not None: self.observer.unschedule(watch) reloader_loops: t.Dict[str, t.Type[ReloaderLoop]] = { "stat": StatReloaderLoop, "watchdog": WatchdogReloaderLoop, } try: __import__("watchdog.observers") except ImportError: reloader_loops["auto"] = reloader_loops["stat"] else: reloader_loops["auto"] = reloader_loops["watchdog"] def ensure_echo_on() -> None: """Ensure that echo mode is enabled. Some tools such as PDB disable it which causes usability issues after a reload.""" # tcgetattr will fail if stdin isn't a tty if sys.stdin is None or not sys.stdin.isatty(): return try: import termios except ImportError: return attributes = termios.tcgetattr(sys.stdin) if not attributes[3] & termios.ECHO: attributes[3] |= termios.ECHO termios.tcsetattr(sys.stdin, termios.TCSANOW, attributes) def run_with_reloader( main_func: t.Callable[[], None], extra_files: t.Optional[t.Iterable[str]] = None, exclude_patterns: t.Optional[t.Iterable[str]] = None, interval: t.Union[int, float] = 1, reloader_type: str = "auto", ) -> None: """Run the given function in an independent Python interpreter.""" import signal signal.signal(signal.SIGTERM, lambda *args: sys.exit(0)) reloader = reloader_loops[reloader_type]( extra_files=extra_files, exclude_patterns=exclude_patterns, interval=interval ) try: if os.environ.get("WERKZEUG_RUN_MAIN") == "true": ensure_echo_on() t = threading.Thread(target=main_func, args=()) t.daemon = True # Enter the reloader to set up initial state, then start # the app thread and reloader update loop. with reloader: t.start() reloader.run() else: sys.exit(reloader.restart_with_reloader()) except KeyboardInterrupt: pass