-
-
Notifications
You must be signed in to change notification settings - Fork 473
/
reload.py
276 lines (245 loc) · 8.04 KB
/
reload.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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
import asyncio
import fnmatch
import logging
import os
import sys
import types
import warnings
from contextlib import contextmanager
from bokeh.application.handlers import CodeHandler
try:
from watchfiles import awatch
except Exception:
async def awatch(*files, stop_event=None):
stop_event = stop_event or asyncio.Event()
modify_times = {}
while not stop_event.is_set():
changes = set()
for path in files:
change = _check_file(path, modify_times)
if change:
changes.add((change, path))
if changes:
yield changes
await asyncio.sleep(0.5)
from ..util import fullpath
from .state import state
_reload_logger = logging.getLogger('panel.io.reload')
_watched_files = set()
_modules = set()
_local_modules = set()
# List of paths to ignore
DEFAULT_FOLDER_DENYLIST = [
"**/.*",
"**/anaconda",
"**/anaconda2",
"**/anaconda3",
"**/dist-packages",
"**/miniconda",
"**/miniconda2",
"**/miniconda3",
"**/node_modules",
"**/pyenv",
"**/site-packages",
"**/venv",
"**/virtualenv",
]
IGNORED_MODULES = [
'bokeh_app',
'geoviews.models.',
'panel.',
'torch.'
]
def in_denylist(filepath):
return any(
file_is_in_folder_glob(filepath, denylisted_folder)
for denylisted_folder in DEFAULT_FOLDER_DENYLIST
)
def file_is_in_folder_glob(filepath, folderpath_glob):
"""
Test whether a file is in some folder with globbing support.
Parameters
----------
filepath : str
A file path.
folderpath_glob: str
A path to a folder that may include globbing.
"""
# Make the glob always end with "/*" so we match files inside subfolders of
# folderpath_glob.
if not folderpath_glob.endswith("*"):
if folderpath_glob.endswith("/"):
folderpath_glob += "*"
else:
folderpath_glob += "/*"
file_dir = os.path.dirname(filepath) + "/"
return fnmatch.fnmatch(file_dir, folderpath_glob)
def watched_modules():
files = list(_watched_files)
module_paths = {}
for module_name in (_modules | _local_modules):
# Some modules play games with sys.modules (e.g. email/__init__.py
# in the standard library), and occasionally this can cause strange
# failures in getattr. Just ignore anything that's not an ordinary
# module.
if module_name not in sys.modules:
continue
module = sys.modules[module_name]
if not isinstance(module, types.ModuleType):
continue
path = getattr(module, "__file__", None)
if not path:
continue
if path.endswith((".pyc", ".pyo")):
path = path[:-1]
path = os.path.abspath(os.path.realpath(path))
module_paths[path] = module_name
files.append(path)
return module_paths, files
async def async_file_watcher(stop_event=None):
while True:
module_paths, files = watched_modules()
async for changes in awatch(*files, stop_event=stop_event):
_reload(module_paths, changes)
await asyncio.sleep(1)
break
if stop_event.is_set():
break
async def setup_autoreload_watcher(stop_event=None):
"""
Installs a periodic callback which checks for changes in watched
files and sys.modules.
"""
try:
import watchfiles # noqa
except Exception:
warnings.warn(
'--autoreload functionality now depends on the watchfiles '
'library. In future versions autoreload will not work without '
'watchfiles being installed. Since it provides a much better '
'user experience consider installing it today.', FutureWarning,
stacklevel=0
)
_reload_logger.debug('Setting up global autoreload watcher.')
await async_file_watcher(stop_event=stop_event)
def watch(filename):
"""
Add a file to the watch list.
"""
_watched_files.add(filename)
def is_subpath(subpath, path):
try:
return os.path.commonpath([path, subpath]) == path
except Exception:
return False
@contextmanager
def record_modules(applications=None, handler=None):
"""
Records modules which are currently imported.
"""
app_paths = set()
if hasattr(handler, '_runner'):
app_paths.add(os.path.dirname(handler._runner.path))
for app in (applications or ()):
if not app._handlers:
continue
for handler in app._handlers:
if isinstance(handler, CodeHandler):
break
else:
continue
if hasattr(handler, '_runner'):
app_paths.add(os.path.dirname(handler._runner.path))
modules = set(sys.modules)
yield
for module_name in set(sys.modules).difference(modules):
if any(module_name.startswith(imodule) for imodule in IGNORED_MODULES):
continue
module = sys.modules[module_name]
try:
spec = getattr(module, "__spec__", None)
if spec is None:
filepath = getattr(module, "__file__", None)
if filepath is None: # no user
continue
else:
filepath = spec.origin
filepath = fullpath(filepath)
if filepath is None or in_denylist(filepath):
continue
if not os.path.isfile(filepath): # e.g. built-in
continue
parent_path = os.path.dirname(filepath)
if any(parent_path == app_path or is_subpath(app_path, parent_path) for app_path in app_paths):
_local_modules.add(module_name)
else:
_modules.add(module_name)
except Exception:
continue
def _reload(module_paths, changes):
"""
Reloads modules depending on the module files that were changed.
Specifically we make a distinction between local modules relative
to the current application paths and global modules. This allows
us to reload the application itself, any local modules imported
by the application or all global modules independently.
"""
_reload_logger.debug('Changes detected by autoreload watcher, unloading modules and reloading sessions.')
local_, global_ = False, False
for _, path in changes:
if path not in module_paths:
continue
module = module_paths[path]
if module in _local_modules and not any(m_.startswith(f'{module}.') for m_ in _modules):
local_ = True
else:
global_ = True
modules_to_delete = set()
if global_:
modules_to_delete |= _modules
if global_ or local_:
modules_to_delete |= _local_modules
for module in modules_to_delete:
if module in sys.modules:
del sys.modules[module]
for doc, loc in state._locations.items():
if not doc.session_context:
continue
elif state._loaded.get(doc):
loc.reload = True
else:
def reload_session(event, loc=loc):
loc.reload = True
doc.on_event('document_ready', reload_session)
def _check_file(path, modify_times):
"""
Checks if a file was modified or deleted and then returns a code,
modeled after watchfiles, indicating the type of change:
- 0: No change
- 2: File modified
- 3: File deleted
Arguments
---------
path: str | os.PathLike
Path of the file to check for modification
modify_times: dict[str, int]
Dictionary of modification times for different paths.
Returns
-------
Status code indicating type of change.
"""
last_modified = modify_times.get(path)
try:
modified = os.stat(path).st_mtime
except FileNotFoundError:
if last_modified:
return 3
return 0
except Exception:
return 0
if last_modified is None:
modify_times[path] = modified
return 0
elif last_modified != modified:
modify_times[path] = modified
return 2