Repository: pypa/pip
File: src/pip/_internal/network/cache.py (Python)

Code before:
        # 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)

                # Change permissions only if there is no risk of following a symlink
                if (
                    os.chmod in os.supports_fd
                    or os.chmod in os.supports_follow_symlinks
                ):
                    os.chmod(
                        path=f.fileno() if os.chmod in os.supports_fd else f.name,
                        mode=(
                            os.stat(self.directory).st_mode
                            & 0o666  # select read/write permissions of cache directory
                            | 0o600  # set owner read/write permissions
                        ),
                        follow_symlinks=os.chmod not in os.supports_follow_symlinks,
                    )

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

Review comment (suggestion):
According to the docs these `os.supports_*` attributes may raise `NotImplementedError`: https://docs.python.org/3/library/os.html#files-and-directories

I think logic to handle this case should be added, pip gets run on a lot of exotic platforms.

Code after:
        # 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)
                # Inherit the read/write permissions of the cache directory
                # to enable multi-user cache use-cases.
                mode = (
                    os.stat(self.directory).st_mode
                    & 0o666  # select read/write permissions of cache directory
                    | 0o600  # set owner read/write permissions
                )
                # Change permissions only if there is no risk of following a symlink.
                if os.chmod in os.supports_fd:
                    os.chmod(f.fileno(), mode)
                elif os.chmod in os.supports_follow_symlinks:
                    os.chmod(f.name, mode, follow_symlinks=False)

            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.