from __future__ import (absolute_import, division, print_function,
unicode_literals)
import json
import os
import sys
from pip._vendor.packaging.requirements import Requirement
from .exceptions import PipToolsError
from .locations import CACHE_DIR
from .utils import as_tuple, key_from_req, lookup_table
class CorruptCacheError(PipToolsError):
def __init__(self, path):
self.path = path
def __str__(self):
lines = [
'The dependency cache seems to have been corrupted.',
'Inspect, or delete, the following file:',
' {}'.format(self.path),
]
return os.linesep.join(lines)
def read_cache_file(cache_file_path):
with open(cache_file_path, 'r') as cache_file:
try:
doc = json.load(cache_file)
except ValueError:
raise CorruptCacheError(cache_file_path)
assert doc['__format__'] == 1, 'Unknown cache file format'
return doc['dependencies']
class DependencyCache(object):
def __init__(self, cache_dir=None):
if cache_dir is None:
cache_dir = CACHE_DIR
if not os.path.isdir(cache_dir):
os.makedirs(cache_dir)
py_version = '.'.join(str(digit) for digit in sys.version_info[:2])
cache_filename = 'depcache-py{}.json'.format(py_version)
self._cache_file = os.path.join(cache_dir, cache_filename)
self._cache = None
@property
def cache(self):
if self._cache is None:
self.read_cache()
return self._cache
def as_cache_key(self, ireq):
name, version, extras = as_tuple(ireq)
if not extras:
extras_string = ""
else:
extras_string = "[{}]".format(",".join(extras))
return name, "{}{}".format(version, extras_string)
def read_cache(self):
if os.path.exists(self._cache_file):
self._cache = read_cache_file(self._cache_file)
else:
self._cache = {}
def write_cache(self):
doc = {
'__format__': 1,
'dependencies': self._cache,
}
with open(self._cache_file, 'w') as f:
json.dump(doc, f, sort_keys=True)
def clear(self):
self._cache = {}
self.write_cache()
def __contains__(self, ireq):
pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
return pkgversion_and_extras in self.cache.get(pkgname, {})
def __getitem__(self, ireq):
pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
return self.cache[pkgname][pkgversion_and_extras]
def __setitem__(self, ireq, values):
pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
self.cache.setdefault(pkgname, {})
self.cache[pkgname][pkgversion_and_extras] = values
self.write_cache()
def get(self, ireq, default=None):
pkgname, pkgversion_and_extras = self.as_cache_key(ireq)
return self.cache.get(pkgname, {}).get(pkgversion_and_extras, default)
def reverse_dependencies(self, ireqs):
ireqs_as_cache_values = [self.as_cache_key(ireq) for ireq in ireqs]
return self._reverse_dependencies(ireqs_as_cache_values)
def _reverse_dependencies(self, cache_keys):
return lookup_table((key_from_req(Requirement(dep_name)), name)
for name, version_and_extras in cache_keys
for dep_name in self.cache[name][version_and_extras])