two version of R2R are hereHEADmaster

1 files changed, 96 insertions, 0 deletions

diff --git a/.venv/lib/python3.12/site-packages/uvicorn/supervisors/watchfilesreload.py b/.venv/lib/python3.12/site-packages/uvicorn/supervisors/watchfilesreload.py
new file mode 100644
index 00000000..e1cb311f
--- /dev/null
+++ b/.venv/lib/python3.12/site-packages/uvicorn/supervisors/watchfilesreload.py
@@ -0,0 +1,96 @@
+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