1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52
|
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()
|