/
watchfilesreload.py
87 lines (73 loc) 路 2.66 KB
/
watchfilesreload.py
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
import logging
from pathlib import Path
from socket import socket
from typing import Callable, List, Optional
from watchfiles import watch
from uvicorn.config import Config
from uvicorn.supervisors.basereload import BaseReload
logger = logging.getLogger("uvicorn.error")
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:
# 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):
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[[Optional[List[socket]]], 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,
)
def should_restart(self) -> Optional[List[Path]]:
changes = next(self.watcher)
unique_paths = {Path(c[1]) for c in changes}
return [p for p in unique_paths if self.watch_filter(p)]