HCAI-Lab/w2-consensus-deepdive-unlearning-artifacts / pyright_venv /lib /python3.9 /site-packages /pip /_internal /exceptions.py
| """Exceptions used throughout package""" | |
| import configparser | |
| from itertools import chain, groupby, repeat | |
| from typing import TYPE_CHECKING, Dict, List, Optional, Union | |
| from pip._vendor.pkg_resources import Distribution | |
| from pip._vendor.requests.models import Request, Response | |
| if TYPE_CHECKING: | |
| from hashlib import _Hash | |
| from pip._internal.metadata import BaseDistribution | |
| from pip._internal.req.req_install import InstallRequirement | |
| class PipError(Exception): | |
| """Base pip exception""" | |
| class ConfigurationError(PipError): | |
| """General exception in configuration""" | |
| class InstallationError(PipError): | |
| """General exception during installation""" | |
| class UninstallationError(PipError): | |
| """General exception during uninstallation""" | |
| class NoneMetadataError(PipError): | |
| """ | |
| Raised when accessing "METADATA" or "PKG-INFO" metadata for a | |
| pip._vendor.pkg_resources.Distribution object and | |
| `dist.has_metadata('METADATA')` returns True but | |
| `dist.get_metadata('METADATA')` returns None (and similarly for | |
| "PKG-INFO"). | |
| """ | |
| def __init__( | |
| self, | |
| dist: Union[Distribution, "BaseDistribution"], | |
| metadata_name: str, | |
| ) -> None: | |
| """ | |
| :param dist: A Distribution object. | |
| :param metadata_name: The name of the metadata being accessed | |
| (can be "METADATA" or "PKG-INFO"). | |
| """ | |
| self.dist = dist | |
| self.metadata_name = metadata_name | |
| def __str__(self) -> str: | |
| # Use `dist` in the error message because its stringification | |
| # includes more information, like the version and location. | |
| return "None {} metadata found for distribution: {}".format( | |
| self.metadata_name, | |
| self.dist, | |
| ) | |
| class UserInstallationInvalid(InstallationError): | |
| """A --user install is requested on an environment without user site.""" | |
| def __str__(self) -> str: | |
| return "User base directory is not specified" | |
| class InvalidSchemeCombination(InstallationError): | |
| def __str__(self) -> str: | |
| before = ", ".join(str(a) for a in self.args[:-1]) | |
| return f"Cannot set {before} and {self.args[-1]} together" | |
| class DistributionNotFound(InstallationError): | |
| """Raised when a distribution cannot be found to satisfy a requirement""" | |
| class RequirementsFileParseError(InstallationError): | |
| """Raised when a general error occurs parsing a requirements file line.""" | |
| class BestVersionAlreadyInstalled(PipError): | |
| """Raised when the most up-to-date version of a package is already | |
| installed.""" | |
| class BadCommand(PipError): | |
| """Raised when virtualenv or a command is not found""" | |
| class CommandError(PipError): | |
| """Raised when there is an error in command-line arguments""" | |
| class PreviousBuildDirError(PipError): | |
| """Raised when there's a previous conflicting build directory""" | |
| class NetworkConnectionError(PipError): | |
| """HTTP connection error""" | |
| def __init__( | |
| self, error_msg: str, response: Response = None, request: Request = None | |
| ) -> None: | |
| """ | |
| Initialize NetworkConnectionError with `request` and `response` | |
| objects. | |
| """ | |
| self.response = response | |
| self.request = request | |
| self.error_msg = error_msg | |
| if ( | |
| self.response is not None | |
| and not self.request | |
| and hasattr(response, "request") | |
| ): | |
| self.request = self.response.request | |
| super().__init__(error_msg, response, request) | |
| def __str__(self) -> str: | |
| return str(self.error_msg) | |
| class InvalidWheelFilename(InstallationError): | |
| """Invalid wheel filename.""" | |
| class UnsupportedWheel(InstallationError): | |
| """Unsupported wheel.""" | |
| class MetadataInconsistent(InstallationError): | |
| """Built metadata contains inconsistent information. | |
| This is raised when the metadata contains values (e.g. name and version) | |
| that do not match the information previously obtained from sdist filename | |
| or user-supplied ``#egg=`` value. | |
| """ | |
| def __init__( | |
| self, ireq: "InstallRequirement", field: str, f_val: str, m_val: str | |
| ) -> None: | |
| self.ireq = ireq | |
| self.field = field | |
| self.f_val = f_val | |
| self.m_val = m_val | |
| def __str__(self) -> str: | |
| template = ( | |
| "Requested {} has inconsistent {}: " | |
| "filename has {!r}, but metadata has {!r}" | |
| ) | |
| return template.format(self.ireq, self.field, self.f_val, self.m_val) | |
| class InstallationSubprocessError(InstallationError): | |
| """A subprocess call failed during installation.""" | |
| def __init__(self, returncode: int, description: str) -> None: | |
| self.returncode = returncode | |
| self.description = description | |
| def __str__(self) -> str: | |
| return ( | |
| "Command errored out with exit status {}: {} " | |
| "Check the logs for full command output." | |
| ).format(self.returncode, self.description) | |
| class HashErrors(InstallationError): | |
| """Multiple HashError instances rolled into one for reporting""" | |
| def __init__(self) -> None: | |
| self.errors: List["HashError"] = [] | |
| def append(self, error: "HashError") -> None: | |
| self.errors.append(error) | |
| def __str__(self) -> str: | |
| lines = [] | |
| self.errors.sort(key=lambda e: e.order) | |
| for cls, errors_of_cls in groupby(self.errors, lambda e: e.__class__): | |
| lines.append(cls.head) | |
| lines.extend(e.body() for e in errors_of_cls) | |
| if lines: | |
| return "\n".join(lines) | |
| return "" | |
| def __bool__(self) -> bool: | |
| return bool(self.errors) | |
| class HashError(InstallationError): | |
| """ | |
| A failure to verify a package against known-good hashes | |
| :cvar order: An int sorting hash exception classes by difficulty of | |
| recovery (lower being harder), so the user doesn't bother fretting | |
| about unpinned packages when he has deeper issues, like VCS | |
| dependencies, to deal with. Also keeps error reports in a | |
| deterministic order. | |
| :cvar head: A section heading for display above potentially many | |
| exceptions of this kind | |
| :ivar req: The InstallRequirement that triggered this error. This is | |
| pasted on after the exception is instantiated, because it's not | |
| typically available earlier. | |
| """ | |
| req: Optional["InstallRequirement"] = None | |
| head = "" | |
| order: int = -1 | |
| def body(self) -> str: | |
| """Return a summary of me for display under the heading. | |
| This default implementation simply prints a description of the | |
| triggering requirement. | |
| :param req: The InstallRequirement that provoked this error, with | |
| its link already populated by the resolver's _populate_link(). | |
| """ | |
| return f" {self._requirement_name()}" | |
| def __str__(self) -> str: | |
| return f"{self.head}\n{self.body()}" | |
| def _requirement_name(self) -> str: | |
| """Return a description of the requirement that triggered me. | |
| This default implementation returns long description of the req, with | |
| line numbers | |
| """ | |
| return str(self.req) if self.req else "unknown package" | |
| class VcsHashUnsupported(HashError): | |
| """A hash was provided for a version-control-system-based requirement, but | |
| we don't have a method for hashing those.""" | |
| order = 0 | |
| head = ( | |
| "Can't verify hashes for these requirements because we don't " | |
| "have a way to hash version control repositories:" | |
| ) | |
| class DirectoryUrlHashUnsupported(HashError): | |
| """A hash was provided for a version-control-system-based requirement, but | |
| we don't have a method for hashing those.""" | |
| order = 1 | |
| head = ( | |
| "Can't verify hashes for these file:// requirements because they " | |
| "point to directories:" | |
| ) | |
| class HashMissing(HashError): | |
| """A hash was needed for a requirement but is absent.""" | |
| order = 2 | |
| head = ( | |
| "Hashes are required in --require-hashes mode, but they are " | |
| "missing from some requirements. Here is a list of those " | |
| "requirements along with the hashes their downloaded archives " | |
| "actually had. Add lines like these to your requirements files to " | |
| "prevent tampering. (If you did not enable --require-hashes " | |
| "manually, note that it turns on automatically when any package " | |
| "has a hash.)" | |
| ) | |
| def __init__(self, gotten_hash: str) -> None: | |
| """ | |
| :param gotten_hash: The hash of the (possibly malicious) archive we | |
| just downloaded | |
| """ | |
| self.gotten_hash = gotten_hash | |
| def body(self) -> str: | |
| # Dodge circular import. | |
| from pip._internal.utils.hashes import FAVORITE_HASH | |
| package = None | |
| if self.req: | |
| # In the case of URL-based requirements, display the original URL | |
| # seen in the requirements file rather than the package name, | |
| # so the output can be directly copied into the requirements file. | |
| package = ( | |
| self.req.original_link | |
| if self.req.original_link | |
| # In case someone feeds something downright stupid | |
| # to InstallRequirement's constructor. | |
| else getattr(self.req, "req", None) | |
| ) | |
| return " {} --hash={}:{}".format( | |
| package or "unknown package", FAVORITE_HASH, self.gotten_hash | |
| ) | |
| class HashUnpinned(HashError): | |
| """A requirement had a hash specified but was not pinned to a specific | |
| version.""" | |
| order = 3 | |
| head = ( | |
| "In --require-hashes mode, all requirements must have their " | |
| "versions pinned with ==. These do not:" | |
| ) | |
| class HashMismatch(HashError): | |
| """ | |
| Distribution file hash values don't match. | |
| :ivar package_name: The name of the package that triggered the hash | |
| mismatch. Feel free to write to this after the exception is raise to | |
| improve its error message. | |
| """ | |
| order = 4 | |
| head = ( | |
| "THESE PACKAGES DO NOT MATCH THE HASHES FROM THE REQUIREMENTS " | |
| "FILE. If you have updated the package versions, please update " | |
| "the hashes. Otherwise, examine the package contents carefully; " | |
| "someone may have tampered with them." | |
| ) | |
| def __init__(self, allowed: Dict[str, List[str]], gots: Dict[str, "_Hash"]) -> None: | |
| """ | |
| :param allowed: A dict of algorithm names pointing to lists of allowed | |
| hex digests | |
| :param gots: A dict of algorithm names pointing to hashes we | |
| actually got from the files under suspicion | |
| """ | |
| self.allowed = allowed | |
| self.gots = gots | |
| def body(self) -> str: | |
| return " {}:\n{}".format(self._requirement_name(), self._hash_comparison()) | |
| def _hash_comparison(self) -> str: | |
| """ | |
| Return a comparison of actual and expected hash values. | |
| Example:: | |
| Expected sha256 abcdeabcdeabcdeabcdeabcdeabcdeabcdeabcdeabcde | |
| or 123451234512345123451234512345123451234512345 | |
| Got bcdefbcdefbcdefbcdefbcdefbcdefbcdefbcdefbcdef | |
| """ | |
| def hash_then_or(hash_name: str) -> "chain[str]": | |
| # For now, all the decent hashes have 6-char names, so we can get | |
| # away with hard-coding space literals. | |
| return chain([hash_name], repeat(" or")) | |
| lines: List[str] = [] | |
| for hash_name, expecteds in self.allowed.items(): | |
| prefix = hash_then_or(hash_name) | |
| lines.extend( | |
| (" Expected {} {}".format(next(prefix), e)) for e in expecteds | |
| ) | |
| lines.append( | |
| " Got {}\n".format(self.gots[hash_name].hexdigest()) | |
| ) | |
| return "\n".join(lines) | |
| class UnsupportedPythonVersion(InstallationError): | |
| """Unsupported python version according to Requires-Python package | |
| metadata.""" | |
| class ConfigurationFileCouldNotBeLoaded(ConfigurationError): | |
| """When there are errors while loading a configuration file""" | |
| def __init__( | |
| self, | |
| reason: str = "could not be loaded", | |
| fname: Optional[str] = None, | |
| error: Optional[configparser.Error] = None, | |
| ) -> None: | |
| super().__init__(error) | |
| self.reason = reason | |
| self.fname = fname | |
| self.error = error | |
| def __str__(self) -> str: | |
| if self.fname is not None: | |
| message_part = f" in {self.fname}." | |
| else: | |
| assert self.error is not None | |
| message_part = f".\n{self.error}\n" | |
| return f"Configuration file {self.reason}{message_part}" | |
Xet Storage Details
- Size:
- 12.8 kB
- Xet hash:
- 8bf0e6524683cd931c7d81ec20ea6e1b7de82c31e4dc606e8c183d4aa32ad18a
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.