| """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: |
| 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 cachable |
| 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 |