Spaces:
No application file
No application file
House_price_prediction
/
BangloreHousePrediction
/.venv
/Lib
/site-packages
/pip
/_internal
/network
/cache.py
"""HTTP cache implementation. | |
""" | |
import os | |
from contextlib import contextmanager | |
from datetime import datetime | |
from typing import BinaryIO, Generator, Optional, Union | |
from pip._vendor.cachecontrol.cache import SeparateBodyBaseCache | |
from pip._vendor.cachecontrol.caches import SeparateBodyFileCache | |
from pip._vendor.requests.models import Response | |
from pip._internal.utils.filesystem import adjacent_tmp_file, replace | |
from pip._internal.utils.misc import ensure_dir | |
def is_from_cache(response: Response) -> bool: | |
return getattr(response, "from_cache", False) | |
def suppressed_cache_errors() -> Generator[None, None, None]: | |
"""If we can't access the cache then we can just skip caching and process | |
requests as if caching wasn't enabled. | |
""" | |
try: | |
yield | |
except OSError: | |
pass | |
class SafeFileCache(SeparateBodyBaseCache): | |
""" | |
A file based cache which is safe to use even when the target directory may | |
not be accessible or writable. | |
There is a race condition when two processes try to write and/or read the | |
same entry at the same time, since each entry consists of two separate | |
files (https://github.com/psf/cachecontrol/issues/324). We therefore have | |
additional logic that makes sure that both files to be present before | |
returning an entry; this fixes the read side of the race condition. | |
For the write side, we assume that the server will only ever return the | |
same data for the same URL, which ought to be the case for files pip is | |
downloading. PyPI does not have a mechanism to swap out a wheel for | |
another wheel, for example. If this assumption is not true, the | |
CacheControl issue will need to be fixed. | |
""" | |
def __init__(self, directory: str) -> None: | |
assert directory is not None, "Cache directory must not be None." | |
super().__init__() | |
self.directory = directory | |
def _get_cache_path(self, name: str) -> str: | |
# From cachecontrol.caches.file_cache.FileCache._fn, brought into our | |
# class for backwards-compatibility and to avoid using a non-public | |
# method. | |
hashed = SeparateBodyFileCache.encode(name) | |
parts = list(hashed[:5]) + [hashed] | |
return os.path.join(self.directory, *parts) | |
def get(self, key: str) -> Optional[bytes]: | |
# The cache entry is only valid if both metadata and body exist. | |
metadata_path = self._get_cache_path(key) | |
body_path = metadata_path + ".body" | |
if not (os.path.exists(metadata_path) and os.path.exists(body_path)): | |
return None | |
with suppressed_cache_errors(): | |
with open(metadata_path, "rb") as f: | |
return f.read() | |
def _write(self, path: str, data: bytes) -> None: | |
with suppressed_cache_errors(): | |
ensure_dir(os.path.dirname(path)) | |
with adjacent_tmp_file(path) as f: | |
f.write(data) | |
replace(f.name, path) | |
def set( | |
self, key: str, value: bytes, expires: Union[int, datetime, None] = None | |
) -> None: | |
path = self._get_cache_path(key) | |
self._write(path, value) | |
def delete(self, key: str) -> None: | |
path = self._get_cache_path(key) | |
with suppressed_cache_errors(): | |
os.remove(path) | |
with suppressed_cache_errors(): | |
os.remove(path + ".body") | |
def get_body(self, key: str) -> Optional[BinaryIO]: | |
# The cache entry is only valid if both metadata and body exist. | |
metadata_path = self._get_cache_path(key) | |
body_path = metadata_path + ".body" | |
if not (os.path.exists(metadata_path) and os.path.exists(body_path)): | |
return None | |
with suppressed_cache_errors(): | |
return open(body_path, "rb") | |
def set_body(self, key: str, body: bytes) -> None: | |
path = self._get_cache_path(key) + ".body" | |
self._write(path, body) | |