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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
|
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) # 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 = []
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
|
