Загрузить файлы в «venv/Lib/site-packages/uvicorn/supervisors»

venv/Lib/site-packages/uvicorn/supervisors/__init__.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from uvicorn.supervisors.basereload import BaseReload
+from uvicorn.supervisors.multiprocess import Multiprocess
+
+if TYPE_CHECKING:
+    ChangeReload: type[BaseReload]
+else:
+    try:
+        from uvicorn.supervisors.watchfilesreload import WatchFilesReload as ChangeReload
+    except ImportError:  # pragma: no cover
+        from uvicorn.supervisors.statreload import StatReload as ChangeReload
+
+__all__ = ["Multiprocess", "ChangeReload"]

venv/Lib/site-packages/uvicorn/supervisors/basereload.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import logging
+import os
+import signal
+import sys
+import threading
+from collections.abc import Callable, Iterator
+from pathlib import Path
+from socket import socket
+from types import FrameType
+
+import click
+
+from uvicorn._subprocess import get_subprocess
+from uvicorn.config import Config
+
+HANDLED_SIGNALS = (
+    signal.SIGINT,  # Unix signal 2. Sent by Ctrl+C.
+    signal.SIGTERM,  # Unix signal 15. Sent by `kill <pid>`.
+)
+
+logger = logging.getLogger("uvicorn.error")
+
+
+class BaseReload:
+    def __init__(
+        self,
+        config: Config,
+        target: Callable[[list[socket] | None], None],
+        sockets: list[socket],
+    ) -> None:
+        self.config = config
+        self.target = target
+        self.sockets = sockets
+        self.should_exit = threading.Event()
+        self.pid = os.getpid()
+        self.is_restarting = False
+        self.reloader_name: str | None = None
+
+    def signal_handler(self, sig: int, frame: FrameType | None) -> None:  # pragma: full coverage
+        """
+        A signal handler that is registered with the parent process.
+        """
+        if sys.platform == "win32" and self.is_restarting:
+            self.is_restarting = False
+        else:
+            self.should_exit.set()
+
+    def run(self) -> None:
+        self.startup()
+        for changes in self:
+            if changes:
+                logger.warning(
+                    "%s detected changes in %s. Reloading...",
+                    self.reloader_name,
+                    ", ".join(map(_display_path, changes)),
+                )
+                self.restart()
+
+        self.shutdown()
+
+    def pause(self) -> None:
+        if self.should_exit.wait(self.config.reload_delay):
+            raise StopIteration()
+
+    def __iter__(self) -> Iterator[list[Path] | None]:
+        return self
+
+    def __next__(self) -> list[Path] | None:
+        return self.should_restart()
+
+    def startup(self) -> None:
+        message = f"Started reloader process [{self.pid}] using {self.reloader_name}"
+        color_message = "Started reloader process [{}] using {}".format(
+            click.style(str(self.pid), fg="cyan", bold=True),
+            click.style(str(self.reloader_name), fg="cyan", bold=True),
+        )
+        logger.info(message, extra={"color_message": color_message})
+
+        for sig in HANDLED_SIGNALS:
+            signal.signal(sig, self.signal_handler)
+
+        self.process = get_subprocess(config=self.config, target=self.target, sockets=self.sockets)
+        self.process.start()
+
+    def restart(self) -> None:
+        if sys.platform == "win32":  # pragma: py-not-win32
+            self.is_restarting = True
+            assert self.process.pid is not None
+            os.kill(self.process.pid, signal.CTRL_C_EVENT)
+
+            # This is a workaround to ensure the Ctrl+C event is processed
+            sys.stdout.write(" ")  # This has to be a non-empty string
+            sys.stdout.flush()
+        else:  # pragma: py-win32
+            self.process.terminate()
+        self.process.join()
+
+        self.process = get_subprocess(config=self.config, target=self.target, sockets=self.sockets)
+        self.process.start()
+
+    def shutdown(self) -> None:
+        if sys.platform == "win32":
+            self.should_exit.set()  # pragma: py-not-win32
+        else:
+            self.process.terminate()  # pragma: py-win32
+        self.process.join()
+
+        for sock in self.sockets:
+            sock.close()
+
+        message = f"Stopping reloader process [{str(self.pid)}]"
+        color_message = "Stopping reloader process [{}]".format(click.style(str(self.pid), fg="cyan", bold=True))
+        logger.info(message, extra={"color_message": color_message})
+
+    def should_restart(self) -> list[Path] | None:
+        raise NotImplementedError("Reload strategies should override should_restart()")
+
+
+def _display_path(path: Path) -> str:
+    try:
+        return f"'{path.relative_to(Path.cwd())}'"
+    except ValueError:
+        return f"'{path}'"

venv/Lib/site-packages/uvicorn/supervisors/statreload.py

@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable, Iterator
+from pathlib import Path
+from socket import socket
+
+from uvicorn.config import Config
+from uvicorn.supervisors.basereload import BaseReload
+
+logger = logging.getLogger("uvicorn.error")
+
+
+class StatReload(BaseReload):
+    def __init__(
+        self,
+        config: Config,
+        target: Callable[[list[socket] | None], None],
+        sockets: list[socket],
+    ) -> None:
+        super().__init__(config, target, sockets)
+        self.reloader_name = "StatReload"
+        self.mtimes: dict[Path, float] = {}
+
+        if config.reload_excludes or config.reload_includes:
+            logger.warning("--reload-include and --reload-exclude have no effect unless watchfiles is installed.")
+
+    def should_restart(self) -> list[Path] | None:
+        self.pause()
+
+        for file in self.iter_py_files():
+            try:
+                mtime = file.stat().st_mtime
+            except OSError:  # pragma: nocover
+                continue
+
+            old_time = self.mtimes.get(file)
+            if old_time is None:
+                self.mtimes[file] = mtime
+                continue
+            elif mtime > old_time:
+                return [file]
+        return None
+
+    def restart(self) -> None:
+        self.mtimes = {}
+        return super().restart()
+
+    def iter_py_files(self) -> Iterator[Path]:
+        for reload_dir in self.config.reload_dirs:
+            for path in list(reload_dir.rglob("*.py")):
+                yield path.resolve()

venv/Lib/site-packages/uvicorn/supervisors/watchfilesreload.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+from collections.abc import Callable
+from pathlib import Path
+from socket import socket
+
+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)  # pragma: full coverage
+        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  # pragma: full coverage
+
+                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  # pragma: full coverage
+
+                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: list[Path] = []
+        for directory in config.reload_dirs:
+            self.reload_dirs.append(directory)
+
+        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,
+            ignore_permission_denied=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