{{ message }}
-
-
Notifications
You must be signed in to change notification settings - Fork 35.4k
Expand file tree
/
Copy pathlinecache.py
More file actions
339 lines (284 loc) · 10.5 KB
/
Copy pathlinecache.py
File metadata and controls
339 lines (284 loc) · 10.5 KB
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
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
"""Cache lines from Python source files.
This is intended to read lines from modules imported -- hence if a filename
is not found, it will look down the module search path for a file by
that name.
"""
__all__ = ["getline", "clearcache", "checkcache", "lazycache"]
# The cache. Maps filenames to either a thunk which will provide source code,
# or a tuple (size, mtime, lines, fullname) once loaded.
cache = {}
_interactive_cache = {}
def clearcache():
"""Clear the cache entirely."""
cache.clear()
def getline(filename, lineno, module_globals=None):
"""Get a line for a Python source file from the cache.
Update the cache if it doesn't contain an entry for this file already."""
lines = getlines(filename, module_globals)
if 1 <= lineno <= len(lines):
return lines[lineno - 1]
return ''
def getlines(filename, module_globals=None):
"""Get the lines for a Python source file from the cache.
Update the cache if it doesn't contain an entry for this file already."""
entry = cache.get(filename, None)
if entry is not None and len(entry) != 1:
return entry[2]
try:
return updatecache(filename, module_globals)
except MemoryError:
clearcache()
return []
def _getline_from_code(filename, lineno):
lines = _getlines_from_code(filename)
if 1 <= lineno <= len(lines):
return lines[lineno - 1]
return ''
def _make_key(code):
return (code.co_filename, code.co_qualname, code.co_firstlineno)
def _getlines_from_code(code):
code_id = _make_key(code)
entry = _interactive_cache.get(code_id, None)
if entry is not None and len(entry) != 1:
return entry[2]
return []
def _source_unavailable(filename):
"""Return True if the source code is unavailable for such file name."""
return (
not filename
or (filename.startswith('<')
and filename.endswith('>')
and not filename.startswith('<frozen '))
)
def checkcache(filename=None):
"""Discard cache entries that are out of date.
(This is not checked upon each call!)"""
if filename is None:
# get keys atomically
filenames = cache.copy().keys()
else:
filenames = [filename]
for filename in filenames:
entry = cache.get(filename, None)
if entry is None or len(entry) == 1:
# lazy cache entry, leave it lazy.
continue
size, mtime, lines, fullname = entry
if mtime is None:
continue # no-op for files loaded via a __loader__
try:
# This import can fail if the interpreter is shutting down
import os
except ImportError:
return
try:
stat = os.stat(fullname)
except (OSError, ValueError):
cache.pop(filename, None)
continue
if size != stat.st_size or mtime != stat.st_mtime:
cache.pop(filename, None)
def updatecache(filename, module_globals=None):
"""Update a cache entry and return its list of lines.
If something's wrong, print a message, discard the cache entry,
and return an empty list."""
# These imports are not at top level because linecache is in the critical
# path of the interpreter startup and importing os and sys take a lot of time
# and slows down the startup sequence.
try:
import os
import sys
import tokenize
except ImportError:
# These import can fail if the interpreter is shutting down
return []
entry = cache.pop(filename, None)
if _source_unavailable(filename):
return []
if filename.startswith('<frozen '):
# This is a frozen module, so we need to use the filename
# from the module globals.
if module_globals is None:
return []
fullname = module_globals.get('__file__')
if fullname is None:
return []
else:
fullname = filename
try:
stat = os.stat(fullname)
except OSError:
basename = filename
# Realise a lazy loader based lookup if there is one
# otherwise try to lookup right now.
lazy_entry = entry if entry is not None and len(entry) == 1 else None
if lazy_entry is None:
lazy_entry = _make_lazycache_entry(filename, module_globals)
data = None
if lazy_entry is not None:
try:
data = lazy_entry[0]()
except (ImportError, OSError):
pass
else:
if data is None:
# No luck, the PEP302 loader cannot find the source
# for this module.
return []
if data is None:
# The file may be inside an archive on the module search path,
# such as a zip file.
try:
data = _read_from_archive(fullname)
except ImportError:
# Can happen if the interpreter is shutting down.
return []
if data is not None:
entry = (
len(data),
None,
[line + '\n' for line in data.splitlines()],
fullname
)
cache[filename] = entry
return entry[2]
# Try looking through the module search path, which is only useful
# when handling a relative filename.
if os.path.isabs(filename):
return []
for dirname in sys.path:
try:
fullname = os.path.join(dirname, basename)
except (TypeError, AttributeError):
# Not sufficiently string-like to do anything useful with.
continue
try:
stat = os.stat(fullname)
break
except (OSError, ValueError):
pass
else:
return []
except ValueError: # may be raised by os.stat()
return []
try:
with tokenize.open(fullname) as fp:
lines = fp.readlines()
except (OSError, UnicodeDecodeError, SyntaxError):
return []
if not lines:
lines = ['\n']
elif not lines[-1].endswith('\n'):
lines[-1] += '\n'
size, mtime = stat.st_size, stat.st_mtime
cache[filename] = size, mtime, lines, fullname
return lines
def _read_from_archive(filename):
"""Return the decoded contents of a file inside an archive on sys.path.
Path entry finders for archives, such as zipimport.zipimporter, have a
get_data() method that reads files by their path below the archive,
which is what __file__ and co_filename contain for modules imported
from it. The archive is one of the parent directories of the file, so
look for a finder registered for one of them. Return None if the file
is not in such an archive.
"""
import os
import sys
importers = sys.path_importer_cache
if importers is None:
# Cleared while the interpreter is shutting down.
return None
path = filename
while True:
parent = os.path.dirname(path)
if parent == path:
return None
path = parent
get_data = getattr(importers.get(path), 'get_data', None)
if get_data is None:
continue
try:
data = get_data(filename)
except (ImportError, OSError):
continue
import importlib.util
try:
return importlib.util.decode_source(data)
except (UnicodeDecodeError, SyntaxError):
return None
def lazycache(filename, module_globals):
"""Seed the cache for filename with module_globals.
The module loader will be asked for the source only when getlines is
called, not immediately.
If there is an entry in the cache already, it is not altered.
:return: True if a lazy load is registered in the cache,
otherwise False. To register such a load a module loader with a
get_source method must be found, the filename must be a cacheable
filename, and the filename must not be already cached.
"""
entry = cache.get(filename, None)
if entry is not None:
return len(entry) == 1
lazy_entry = _make_lazycache_entry(filename, module_globals)
if lazy_entry is not None:
cache[filename] = lazy_entry
return True
return False
def _make_lazycache_entry(filename, module_globals):
if not filename or (filename.startswith('<') and filename.endswith('>')):
return None
if module_globals is not None and not isinstance(module_globals, dict):
raise TypeError(f'module_globals must be a dict, not {type(module_globals).__qualname__}')
if not module_globals or '__name__' not in module_globals:
return None
spec = module_globals.get('__spec__')
name = getattr(spec, 'name', None) or module_globals['__name__']
if name is None:
return None
loader = _bless_my_loader(module_globals)
if loader is None:
return None
get_source = getattr(loader, 'get_source', None)
if get_source is None:
return None
def get_lines(name=name, *args, **kwargs):
return get_source(name, *args, **kwargs)
return (get_lines,)
def _bless_my_loader(module_globals):
# Similar to _bless_my_loader() in importlib._bootstrap_external,
# but always emits warnings instead of errors.
loader = module_globals.get('__loader__')
if loader is None and '__spec__' not in module_globals:
return None
spec = module_globals.get('__spec__')
# The __main__ module has __spec__ = None.
if spec is None and module_globals.get('__name__') == '__main__':
return loader
spec_loader = getattr(spec, 'loader', None)
if spec_loader is None:
import warnings
warnings.warn(
'Module globals is missing a __spec__.loader',
DeprecationWarning)
return loader
assert spec_loader is not None
if loader is not None and loader != spec_loader:
import warnings
warnings.warn(
'Module globals; __loader__ != __spec__.loader',
DeprecationWarning)
return loader
return spec_loader
def _register_code(code, string, name):
entry = (len(string),
None,
[line + '\n' for line in string.splitlines()],
name)
stack = [code]
while stack:
code = stack.pop()
for const in code.co_consts:
if isinstance(const, type(code)):
stack.append(const)
key = _make_key(code)
_interactive_cache[key] = entry
You can’t perform that action at this time.