123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182 |
- """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.
- """
- import functools
- import sys
- import os
- import tokenize
- __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 = {}
- 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."""
- if filename in cache:
- entry = cache[filename]
- if len(entry) != 1:
- return cache[filename][2]
- try:
- return updatecache(filename, module_globals)
- except MemoryError:
- clearcache()
- return []
- def checkcache(filename=None):
- """Discard cache entries that are out of date.
- (This is not checked upon each call!)"""
- if filename is None:
- filenames = list(cache.keys())
- elif filename in cache:
- filenames = [filename]
- else:
- return
- for filename in filenames:
- entry = cache[filename]
- if 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:
- stat = os.stat(fullname)
- except OSError:
- 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."""
- if filename in cache:
- if len(cache[filename]) != 1:
- cache.pop(filename, None)
- if not filename or (filename.startswith('<') and filename.endswith('>')):
- return []
- 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.
- if lazycache(filename, module_globals):
- try:
- data = cache[filename][0]()
- except (ImportError, OSError):
- pass
- else:
- if data is None:
- # No luck, the PEP302 loader cannot find the source
- # for this module.
- return []
- cache[filename] = (
- len(data),
- None,
- [line + '\n' for line in data.splitlines()],
- fullname
- )
- return cache[filename][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:
- pass
- else:
- return []
- try:
- with tokenize.open(fullname) as fp:
- lines = fp.readlines()
- except (OSError, UnicodeDecodeError, SyntaxError):
- return []
- if lines and 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 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.
- """
- if filename in cache:
- if len(cache[filename]) == 1:
- return True
- else:
- return False
- if not filename or (filename.startswith('<') and filename.endswith('>')):
- return False
- # Try for a __loader__, if available
- if module_globals and '__name__' in module_globals:
- name = module_globals['__name__']
- if (loader := module_globals.get('__loader__')) is None:
- if spec := module_globals.get('__spec__'):
- try:
- loader = spec.loader
- except AttributeError:
- pass
- get_source = getattr(loader, 'get_source', None)
- if name and get_source:
- get_lines = functools.partial(get_source, name)
- cache[filename] = (get_lines,)
- return True
- return False
|