diff options
author | S. Solomon Darnell | 2025-03-28 21:52:21 -0500 |
---|---|---|
committer | S. Solomon Darnell | 2025-03-28 21:52:21 -0500 |
commit | 4a52a71956a8d46fcb7294ac71734504bb09bcc2 (patch) | |
tree | ee3dc5af3b6313e921cd920906356f5d4febc4ed /.venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py | |
parent | cc961e04ba734dd72309fb548a2f97d67d578813 (diff) | |
download | gn-ai-master.tar.gz |
Diffstat (limited to '.venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py')
-rw-r--r-- | .venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py | 55 |
1 files changed, 55 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py b/.venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py new file mode 100644 index 00000000..2e25dd4a --- /dev/null +++ b/.venv/lib/python3.12/site-packages/uvicorn/supervisors/statreload.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import logging +from pathlib import Path +from socket import socket +from typing import Callable, Iterator + +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() |