"""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)

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 []

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 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