[config.git] / djavu-asus / elpy / rpc-venv / lib / python3.11 / site-packages / pip / _internal / cache.py

"""Cache Management
"""

import hashlib
import json
import logging
import os
from pathlib import Path
from typing import Any, Dict, List, Optional, Set

from pip._vendor.packaging.tags import Tag, interpreter_name, interpreter_version
from pip._vendor.packaging.utils import canonicalize_name

from pip._internal.exceptions import InvalidWheelFilename
from pip._internal.models.direct_url import DirectUrl
from pip._internal.models.format_control import FormatControl
from pip._internal.models.link import Link
from pip._internal.models.wheel import Wheel
from pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds
from pip._internal.utils.urls import path_to_url

logger = logging.getLogger(__name__)

ORIGIN_JSON_NAME = "origin.json"


def _hash_dict(d: Dict[str, str]) -> str:
    """Return a stable sha224 of a dictionary."""
    s = json.dumps(d, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
    return hashlib.sha224(s.encode("ascii")).hexdigest()


class Cache:
    """An abstract class - provides cache directories for data from links


    :param cache_dir: The root of the cache.
    :param format_control: An object of FormatControl class to limit
        binaries being read from the cache.
    :param allowed_formats: which formats of files the cache should store.
        ('binary' and 'source' are the only allowed values)
    """

    def __init__(
        self, cache_dir: str, format_control: FormatControl, allowed_formats: Set[str]
    ) -> None:
        super().__init__()
        assert not cache_dir or os.path.isabs(cache_dir)
        self.cache_dir = cache_dir or None
        self.format_control = format_control
        self.allowed_formats = allowed_formats

        _valid_formats = {"source", "binary"}
        assert self.allowed_formats.union(_valid_formats) == _valid_formats

    def _get_cache_path_parts(self, link: Link) -> List[str]:
        """Get parts of part that must be os.path.joined with cache_dir"""

        # We want to generate an url to use as our cache key, we don't want to
        # just re-use the URL because it might have other items in the fragment
        # and we don't care about those.
        key_parts = {"url": link.url_without_fragment}
        if link.hash_name is not None and link.hash is not None:
            key_parts[link.hash_name] = link.hash
        if link.subdirectory_fragment:
            key_parts["subdirectory"] = link.subdirectory_fragment

        # Include interpreter name, major and minor version in cache key
        # to cope with ill-behaved sdists that build a different wheel
        # depending on the python version their setup.py is being run on,
        # and don't encode the difference in compatibility tags.
        # https://github.com/pypa/pip/issues/7296
        key_parts["interpreter_name"] = interpreter_name()
        key_parts["interpreter_version"] = interpreter_version()

        # Encode our key url with sha224, we'll use this because it has similar
        # security properties to sha256, but with a shorter total output (and
        # thus less secure). However the differences don't make a lot of
        # difference for our use case here.
        hashed = _hash_dict(key_parts)

        # We want to nest the directories some to prevent having a ton of top
        # level directories where we might run out of sub directories on some
        # FS.
        parts = [hashed[:2], hashed[2:4], hashed[4:6], hashed[6:]]

        return parts

    def _get_candidates(self, link: Link, canonical_package_name: str) -> List[Any]:
        can_not_cache = not self.cache_dir or not canonical_package_name or not link
        if can_not_cache:
            return []

        formats = self.format_control.get_allowed_formats(canonical_package_name)
        if not self.allowed_formats.intersection(formats):
            return []

        candidates = []
        path = self.get_path_for_link(link)
        if os.path.isdir(path):
            for candidate in os.listdir(path):
                candidates.append((candidate, path))
        return candidates

    def get_path_for_link(self, link: Link) -> str:
        """Return a directory to store cached items in for link."""
        raise NotImplementedError()

    def get(
        self,
        link: Link,
        package_name: Optional[str],
        supported_tags: List[Tag],
    ) -> Link:
        """Returns a link to a cached item if it exists, otherwise returns the
        passed link.
        """
        raise NotImplementedError()


class SimpleWheelCache(Cache):
    """A cache of wheels for future installs."""

    def __init__(self, cache_dir: str, format_control: FormatControl) -> None:
        super().__init__(cache_dir, format_control, {"binary"})

    def get_path_for_link(self, link: Link) -> str:
        """Return a directory to store cached wheels for link

        Because there are M wheels for any one sdist, we provide a directory
        to cache them in, and then consult that directory when looking up
        cache hits.

        We only insert things into the cache if they have plausible version
        numbers, so that we don't contaminate the cache with things that were
        not unique. E.g. ./package might have dozens of installs done for it
        and build a version of 0.0...and if we built and cached a wheel, we'd
        end up using the same wheel even if the source has been edited.

        :param link: The link of the sdist for which this will cache wheels.
        """
        parts = self._get_cache_path_parts(link)
        assert self.cache_dir
        # Store wheels within the root cache_dir
        return os.path.join(self.cache_dir, "wheels", *parts)

    def get(
        self,
        link: Link,
        package_name: Optional[str],
        supported_tags: List[Tag],
    ) -> Link:
        candidates = []

        if not package_name:
            return link

        canonical_package_name = canonicalize_name(package_name)
        for wheel_name, wheel_dir in self._get_candidates(link, canonical_package_name):
            try:
                wheel = Wheel(wheel_name)
            except InvalidWheelFilename:
                continue
            if canonicalize_name(wheel.name) != canonical_package_name:
                logger.debug(
                    "Ignoring cached wheel %s for %s as it "
                    "does not match the expected distribution name %s.",
                    wheel_name,
                    link,
                    package_name,
                )
                continue
            if not wheel.supported(supported_tags):
                # Built for a different python/arch/etc
                continue
            candidates.append(
                (
                    wheel.support_index_min(supported_tags),
                    wheel_name,
                    wheel_dir,
                )
            )

        if not candidates:
            return link

        _, wheel_name, wheel_dir = min(candidates)
        return Link(path_to_url(os.path.join(wheel_dir, wheel_name)))


class EphemWheelCache(SimpleWheelCache):
    """A SimpleWheelCache that creates it's own temporary cache directory"""

    def __init__(self, format_control: FormatControl) -> None:
        self._temp_dir = TempDirectory(
            kind=tempdir_kinds.EPHEM_WHEEL_CACHE,
            globally_managed=True,
        )

        super().__init__(self._temp_dir.path, format_control)


class CacheEntry:
    def __init__(
        self,
        link: Link,
        persistent: bool,
    ):
        self.link = link
        self.persistent = persistent
        self.origin: Optional[DirectUrl] = None
        origin_direct_url_path = Path(self.link.file_path).parent / ORIGIN_JSON_NAME
        if origin_direct_url_path.exists():
            self.origin = DirectUrl.from_json(origin_direct_url_path.read_text())


class WheelCache(Cache):
    """Wraps EphemWheelCache and SimpleWheelCache into a single Cache

    This Cache allows for gracefully degradation, using the ephem wheel cache
    when a certain link is not found in the simple wheel cache first.
    """

    def __init__(
        self, cache_dir: str, format_control: Optional[FormatControl] = None
    ) -> None:
        if format_control is None:
            format_control = FormatControl()
        super().__init__(cache_dir, format_control, {"binary"})
        self._wheel_cache = SimpleWheelCache(cache_dir, format_control)
        self._ephem_cache = EphemWheelCache(format_control)

    def get_path_for_link(self, link: Link) -> str:
        return self._wheel_cache.get_path_for_link(link)

    def get_ephem_path_for_link(self, link: Link) -> str:
        return self._ephem_cache.get_path_for_link(link)

    def get(
        self,
        link: Link,
        package_name: Optional[str],
        supported_tags: List[Tag],
    ) -> Link:
        cache_entry = self.get_cache_entry(link, package_name, supported_tags)
        if cache_entry is None:
            return link
        return cache_entry.link

    def get_cache_entry(
        self,
        link: Link,
        package_name: Optional[str],
        supported_tags: List[Tag],
    ) -> Optional[CacheEntry]:
        """Returns a CacheEntry with a link to a cached item if it exists or
        None. The cache entry indicates if the item was found in the persistent
        or ephemeral cache.
        """
        retval = self._wheel_cache.get(
            link=link,
            package_name=package_name,
            supported_tags=supported_tags,
        )
        if retval is not link:
            return CacheEntry(retval, persistent=True)

        retval = self._ephem_cache.get(
            link=link,
            package_name=package_name,
            supported_tags=supported_tags,
        )
        if retval is not link:
            return CacheEntry(retval, persistent=False)

        return None

    @staticmethod
    def record_download_origin(cache_dir: str, download_info: DirectUrl) -> None:
        origin_path = Path(cache_dir) / ORIGIN_JSON_NAME
        if origin_path.is_file():
            origin = DirectUrl.from_json(origin_path.read_text())
            # TODO: use DirectUrl.equivalent when https://github.com/pypa/pip/pull/10564
            # is merged.
            if origin.url != download_info.url:
                logger.warning(
                    "Origin URL %s in cache entry %s does not match download URL %s. "
                    "This is likely a pip bug or a cache corruption issue.",
                    origin.url,
                    cache_dir,
                    download_info.url,
                )
        origin_path.write_text(download_info.to_json(), encoding="utf-8")
Commit	Line	Data
53e6db90 DC	1	"""Cache Management
	2	"""
	3
	4	import hashlib
	5	import json
	6	import logging
	7	import os
	8	from pathlib import Path
	9	from typing import Any, Dict, List, Optional, Set
	10
	11	from pip._vendor.packaging.tags import Tag, interpreter_name, interpreter_version
	12	from pip._vendor.packaging.utils import canonicalize_name
	13
	14	from pip._internal.exceptions import InvalidWheelFilename
	15	from pip._internal.models.direct_url import DirectUrl
	16	from pip._internal.models.format_control import FormatControl
	17	from pip._internal.models.link import Link
	18	from pip._internal.models.wheel import Wheel
	19	from pip._internal.utils.temp_dir import TempDirectory, tempdir_kinds
	20	from pip._internal.utils.urls import path_to_url
	21
	22	logger = logging.getLogger(__name__)
	23
	24	ORIGIN_JSON_NAME = "origin.json"
	25
	26
	27	def _hash_dict(d: Dict[str, str]) -> str:
	28	"""Return a stable sha224 of a dictionary."""
	29	s = json.dumps(d, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
	30	return hashlib.sha224(s.encode("ascii")).hexdigest()
	31
	32
	33	class Cache:
	34	"""An abstract class - provides cache directories for data from links
	35
	36
	37	:param cache_dir: The root of the cache.
	38	:param format_control: An object of FormatControl class to limit
	39	binaries being read from the cache.
	40	:param allowed_formats: which formats of files the cache should store.
	41	('binary' and 'source' are the only allowed values)
	42	"""
	43
	44	def __init__(
	45	self, cache_dir: str, format_control: FormatControl, allowed_formats: Set[str]
	46	) -> None:
	47	super().__init__()
	48	assert not cache_dir or os.path.isabs(cache_dir)
	49	self.cache_dir = cache_dir or None
	50	self.format_control = format_control
	51	self.allowed_formats = allowed_formats
	52
	53	_valid_formats = {"source", "binary"}
	54	assert self.allowed_formats.union(_valid_formats) == _valid_formats
	55
	56	def _get_cache_path_parts(self, link: Link) -> List[str]:
	57	"""Get parts of part that must be os.path.joined with cache_dir"""
	58
	59	# We want to generate an url to use as our cache key, we don't want to
	60	# just re-use the URL because it might have other items in the fragment
	61	# and we don't care about those.
	62	key_parts = {"url": link.url_without_fragment}
	63	if link.hash_name is not None and link.hash is not None:
	64	key_parts[link.hash_name] = link.hash
65	if link.subdirectory_fragment:
66	key_parts["subdirectory"] = link.subdirectory_fragment
67
68	# Include interpreter name, major and minor version in cache key
69	# to cope with ill-behaved sdists that build a different wheel
70	# depending on the python version their setup.py is being run on,
71	# and don't encode the difference in compatibility tags.
72	# https://github.com/pypa/pip/issues/7296
73	key_parts["interpreter_name"] = interpreter_name()
74	key_parts["interpreter_version"] = interpreter_version()
75
76	# Encode our key url with sha224, we'll use this because it has similar
77	# security properties to sha256, but with a shorter total output (and
78	# thus less secure). However the differences don't make a lot of
79	# difference for our use case here.
80	hashed = _hash_dict(key_parts)
81
82	# We want to nest the directories some to prevent having a ton of top
83	# level directories where we might run out of sub directories on some
84	# FS.
85	parts = [hashed[:2], hashed[2:4], hashed[4:6], hashed[6:]]
86
87	return parts
88
89	def _get_candidates(self, link: Link, canonical_package_name: str) -> List[Any]:
90	can_not_cache = not self.cache_dir or not canonical_package_name or not link
91	if can_not_cache:
92	return []
93
94	formats = self.format_control.get_allowed_formats(canonical_package_name)
95	if not self.allowed_formats.intersection(formats):
96	return []
97
98	candidates = []
99	path = self.get_path_for_link(link)
100	if os.path.isdir(path):
101	for candidate in os.listdir(path):
102	candidates.append((candidate, path))
103	return candidates
104
105	def get_path_for_link(self, link: Link) -> str:
106	"""Return a directory to store cached items in for link."""
107	raise NotImplementedError()
108
109	def get(
110	self,
111	link: Link,
112	package_name: Optional[str],
113	supported_tags: List[Tag],
114	) -> Link:
115	"""Returns a link to a cached item if it exists, otherwise returns the
116	passed link.
117	"""
118	raise NotImplementedError()
119
120
121	class SimpleWheelCache(Cache):
122	"""A cache of wheels for future installs."""
123
124	def __init__(self, cache_dir: str, format_control: FormatControl) -> None:
125	super().__init__(cache_dir, format_control, {"binary"})
126
127	def get_path_for_link(self, link: Link) -> str:
128	"""Return a directory to store cached wheels for link
129
130	Because there are M wheels for any one sdist, we provide a directory
131	to cache them in, and then consult that directory when looking up
132	cache hits.
133
134	We only insert things into the cache if they have plausible version
135	numbers, so that we don't contaminate the cache with things that were
136	not unique. E.g. ./package might have dozens of installs done for it
137	and build a version of 0.0...and if we built and cached a wheel, we'd
138	end up using the same wheel even if the source has been edited.
139
140	:param link: The link of the sdist for which this will cache wheels.
141	"""
142	parts = self._get_cache_path_parts(link)
143	assert self.cache_dir
144	# Store wheels within the root cache_dir
145	return os.path.join(self.cache_dir, "wheels", *parts)
146
147	def get(
148	self,
149	link: Link,
150	package_name: Optional[str],
151	supported_tags: List[Tag],
152	) -> Link:
153	candidates = []
154
155	if not package_name:
156	return link
157
158	canonical_package_name = canonicalize_name(package_name)
159	for wheel_name, wheel_dir in self._get_candidates(link, canonical_package_name):
160	try:
161	wheel = Wheel(wheel_name)
162	except InvalidWheelFilename:
163	continue
164	if canonicalize_name(wheel.name) != canonical_package_name:
165	logger.debug(
166	"Ignoring cached wheel %s for %s as it "
167	"does not match the expected distribution name %s.",
168	wheel_name,
169	link,
170	package_name,
171	)
172	continue
173	if not wheel.supported(supported_tags):
174	# Built for a different python/arch/etc
175	continue
176	candidates.append(
177	(
178	wheel.support_index_min(supported_tags),
179	wheel_name,
180	wheel_dir,
181	)
182	)
183
184	if not candidates:
185	return link
186
187	_, wheel_name, wheel_dir = min(candidates)
188	return Link(path_to_url(os.path.join(wheel_dir, wheel_name)))
189
190
191	class EphemWheelCache(SimpleWheelCache):
192	"""A SimpleWheelCache that creates it's own temporary cache directory"""
193
194	def __init__(self, format_control: FormatControl) -> None:
195	self._temp_dir = TempDirectory(
196	kind=tempdir_kinds.EPHEM_WHEEL_CACHE,
197	globally_managed=True,
198	)
199
200	super().__init__(self._temp_dir.path, format_control)
201
202
203	class CacheEntry:
204	def __init__(
205	self,
206	link: Link,
207	persistent: bool,
208	):
209	self.link = link
210	self.persistent = persistent
211	self.origin: Optional[DirectUrl] = None
212	origin_direct_url_path = Path(self.link.file_path).parent / ORIGIN_JSON_NAME
213	if origin_direct_url_path.exists():
214	self.origin = DirectUrl.from_json(origin_direct_url_path.read_text())
215
216
217	class WheelCache(Cache):
218	"""Wraps EphemWheelCache and SimpleWheelCache into a single Cache
219
220	This Cache allows for gracefully degradation, using the ephem wheel cache
221	when a certain link is not found in the simple wheel cache first.
222	"""
223
224	def __init__(
225	self, cache_dir: str, format_control: Optional[FormatControl] = None
226	) -> None:
227	if format_control is None:
228	format_control = FormatControl()
229	super().__init__(cache_dir, format_control, {"binary"})
230	self._wheel_cache = SimpleWheelCache(cache_dir, format_control)
231	self._ephem_cache = EphemWheelCache(format_control)
232
233	def get_path_for_link(self, link: Link) -> str:
234	return self._wheel_cache.get_path_for_link(link)
235
236	def get_ephem_path_for_link(self, link: Link) -> str:
237	return self._ephem_cache.get_path_for_link(link)
238
239	def get(
240	self,
241	link: Link,
242	package_name: Optional[str],
243	supported_tags: List[Tag],
244	) -> Link:
245	cache_entry = self.get_cache_entry(link, package_name, supported_tags)
246	if cache_entry is None:
247	return link
248	return cache_entry.link
249
250	def get_cache_entry(
251	self,
252	link: Link,
253	package_name: Optional[str],
254	supported_tags: List[Tag],
255	) -> Optional[CacheEntry]:
256	"""Returns a CacheEntry with a link to a cached item if it exists or
257	None. The cache entry indicates if the item was found in the persistent
258	or ephemeral cache.
259	"""
260	retval = self._wheel_cache.get(
261	link=link,
262	package_name=package_name,
263	supported_tags=supported_tags,
264	)
265	if retval is not link:
266	return CacheEntry(retval, persistent=True)
267
268	retval = self._ephem_cache.get(
269	link=link,
270	package_name=package_name,
271	supported_tags=supported_tags,
272	)
273	if retval is not link:
274	return CacheEntry(retval, persistent=False)
275
276	return None
277
278	@staticmethod
279	def record_download_origin(cache_dir: str, download_info: DirectUrl) -> None:
280	origin_path = Path(cache_dir) / ORIGIN_JSON_NAME
281	if origin_path.is_file():
282	origin = DirectUrl.from_json(origin_path.read_text())
283	# TODO: use DirectUrl.equivalent when https://github.com/pypa/pip/pull/10564
284	# is merged.
285	if origin.url != download_info.url:
286	logger.warning(
287	"Origin URL %s in cache entry %s does not match download URL %s. "
288	"This is likely a pip bug or a cache corruption issue.",
289	origin.url,
290	cache_dir,
291	download_info.url,
292	)
293	origin_path.write_text(download_info.to_json(), encoding="utf-8")