Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2026-03-21 15:45
📄 __init__.py """A package that contains models that represent entities."""	62 B	2026-01-20 12:58
📁 __pycache__	-	2026-03-21 15:45
📄 candidate.py from dataclasses import dataclass from pip._vendor.packaging.version import Version from pip._vendor.packaging.version import parse as parse_version from pip._internal.models.link import Link @dataclass(frozen=True) class InstallationCandidate: """Represents a potential "candidate" for installation.""" __slots__ = ["name", "version", "link"] name: str version: Version link: Link def __init__(self, name: str, version: str, link: Link) -> None: object.__setattr__(self, "name", name) object.__setattr__(self, "version", parse_version(version)) object.__setattr__(self, "link", link) def __str__(self) -> str: return f"{self.name!r} candidate (version {self.version} at {self.link})"	753 B	2026-01-20 12:58
📄 direct_url.py """PEP 610""" from __future__ import annotations import json import re import urllib.parse from collections.abc import Iterable from dataclasses import dataclass from typing import Any, ClassVar, TypeVar, Union __all__ = [ "DirectUrl", "DirectUrlValidationError", "DirInfo", "ArchiveInfo", "VcsInfo", ] T = TypeVar("T") DIRECT_URL_METADATA_NAME = "direct_url.json" ENV_VAR_RE = re.compile(r"^\$\{[A-Za-z0-9-_]+\}(:\$\{[A-Za-z0-9-_]+\})?$") class DirectUrlValidationError(Exception): pass def _get( d: dict[str, Any], expected_type: type[T], key: str, default: T \| None = None ) -> T \| None: """Get value from dictionary and verify expected type.""" if key not in d: return default value = d[key] if not isinstance(value, expected_type): raise DirectUrlValidationError( f"{value!r} has unexpected type for {key} (expected {expected_type})" ) return value def _get_required( d: dict[str, Any], expected_type: type[T], key: str, default: T \| None = None ) -> T: value = _get(d, expected_type, key, default) if value is None: raise DirectUrlValidationError(f"{key} must have a value") return value def _exactly_one_of(infos: Iterable[InfoType \| None]) -> InfoType: infos = [info for info in infos if info is not None] if not infos: raise DirectUrlValidationError( "missing one of archive_info, dir_info, vcs_info" ) if len(infos) > 1: raise DirectUrlValidationError( "more than one of archive_info, dir_info, vcs_info" ) assert infos[0] is not None return infos[0] def _filter_none(**kwargs: Any) -> dict[str, Any]: """Make dict excluding None values.""" return {k: v for k, v in kwargs.items() if v is not None} @dataclass class VcsInfo: name: ClassVar = "vcs_info" vcs: str commit_id: str requested_revision: str \| None = None @classmethod def _from_dict(cls, d: dict[str, Any] \| None) -> VcsInfo \| None: if d is None: return None return cls( vcs=_get_required(d, str, "vcs"), commit_id=_get_required(d, str, "commit_id"), requested_revision=_get(d, str, "requested_revision"), ) def _to_dict(self) -> dict[str, Any]: return _filter_none( vcs=self.vcs, requested_revision=self.requested_revision, commit_id=self.commit_id, ) class ArchiveInfo: name = "archive_info" def __init__( self, hash: str \| None = None, hashes: dict[str, str] \| None = None, ) -> None: # set hashes before hash, since the hash setter will further populate hashes self.hashes = hashes self.hash = hash @property def hash(self) -> str \| None: return self._hash @hash.setter def hash(self, value: str \| None) -> None: if value is not None: # Auto-populate the hashes key to upgrade to the new format automatically. # We don't back-populate the legacy hash key from hashes. try: hash_name, hash_value = value.split("=", 1) except ValueError: raise DirectUrlValidationError( f"invalid archive_info.hash format: {value!r}" ) if self.hashes is None: self.hashes = {hash_name: hash_value} elif hash_name not in self.hashes: self.hashes = self.hashes.copy() self.hashes[hash_name] = hash_value self._hash = value @classmethod def _from_dict(cls, d: dict[str, Any] \| None) -> ArchiveInfo \| None: if d is None: return None return cls(hash=_get(d, str, "hash"), hashes=_get(d, dict, "hashes")) def _to_dict(self) -> dict[str, Any]: return _filter_none(hash=self.hash, hashes=self.hashes) @dataclass class DirInfo: name: ClassVar = "dir_info" editable: bool = False @classmethod def _from_dict(cls, d: dict[str, Any] \| None) -> DirInfo \| None: if d is None: return None return cls(editable=_get_required(d, bool, "editable", default=False)) def _to_dict(self) -> dict[str, Any]: return _filter_none(editable=self.editable or None) InfoType = Union[ArchiveInfo, DirInfo, VcsInfo] @dataclass class DirectUrl: url: str info: InfoType subdirectory: str \| None = None def _remove_auth_from_netloc(self, netloc: str) -> str: if "@" not in netloc: return netloc user_pass, netloc_no_user_pass = netloc.split("@", 1) if ( isinstance(self.info, VcsInfo) and self.info.vcs == "git" and user_pass == "git" ): return netloc if ENV_VAR_RE.match(user_pass): return netloc return netloc_no_user_pass @property def redacted_url(self) -> str: """url with user:password part removed unless it is formed with environment variables as specified in PEP 610, or it is ``git`` in the case of a git URL. """ purl = urllib.parse.urlsplit(self.url) netloc = self._remove_auth_from_netloc(purl.netloc) surl = urllib.parse.urlunsplit( (purl.scheme, netloc, purl.path, purl.query, purl.fragment) ) return surl def validate(self) -> None: self.from_dict(self.to_dict()) @classmethod def from_dict(cls, d: dict[str, Any]) -> DirectUrl: return DirectUrl( url=_get_required(d, str, "url"), subdirectory=_get(d, str, "subdirectory"), info=_exactly_one_of( [ ArchiveInfo._from_dict(_get(d, dict, "archive_info")), DirInfo._from_dict(_get(d, dict, "dir_info")), VcsInfo._from_dict(_get(d, dict, "vcs_info")), ] ), ) def to_dict(self) -> dict[str, Any]: res = _filter_none( url=self.redacted_url, subdirectory=self.subdirectory, ) res[self.info.name] = self.info._to_dict() return res @classmethod def from_json(cls, s: str) -> DirectUrl: return cls.from_dict(json.loads(s)) def to_json(self) -> str: return json.dumps(self.to_dict(), sort_keys=True) def is_local_editable(self) -> bool: return isinstance(self.info, DirInfo) and self.info.editable	6.4 KB	2026-01-20 12:58
📄 format_control.py from __future__ import annotations from pip._vendor.packaging.utils import canonicalize_name from pip._internal.exceptions import CommandError class FormatControl: """Helper for managing formats from which a package can be installed.""" __slots__ = ["no_binary", "only_binary"] def __init__( self, no_binary: set[str] \| None = None, only_binary: set[str] \| None = None, ) -> None: if no_binary is None: no_binary = set() if only_binary is None: only_binary = set() self.no_binary = no_binary self.only_binary = only_binary def __eq__(self, other: object) -> bool: if not isinstance(other, self.__class__): return NotImplemented if self.__slots__ != other.__slots__: return False return all(getattr(self, k) == getattr(other, k) for k in self.__slots__) def __repr__(self) -> str: return f"{self.__class__.__name__}({self.no_binary}, {self.only_binary})" @staticmethod def handle_mutual_excludes(value: str, target: set[str], other: set[str]) -> None: if value.startswith("-"): raise CommandError( "--no-binary / --only-binary option requires 1 argument." ) new = value.split(",") while ":all:" in new: other.clear() target.clear() target.add(":all:") del new[: new.index(":all:") + 1] # Without a none, we want to discard everything as :all: covers it if ":none:" not in new: return for name in new: if name == ":none:": target.clear() continue name = canonicalize_name(name) other.discard(name) target.add(name) def get_allowed_formats(self, canonical_name: str) -> frozenset[str]: result = {"binary", "source"} if canonical_name in self.only_binary: result.discard("source") elif canonical_name in self.no_binary: result.discard("binary") elif ":all:" in self.only_binary: result.discard("source") elif ":all:" in self.no_binary: result.discard("binary") return frozenset(result) def disallow_binaries(self) -> None: self.handle_mutual_excludes( ":all:", self.no_binary, self.only_binary, )	2.41 KB	2026-01-20 12:58
📄 index.py import urllib.parse class PackageIndex: """Represents a Package Index and provides easier access to endpoints""" __slots__ = ["url", "netloc", "simple_url", "pypi_url", "file_storage_domain"] def __init__(self, url: str, file_storage_domain: str) -> None: super().__init__() self.url = url self.netloc = urllib.parse.urlsplit(url).netloc self.simple_url = self._url_for_path("simple") self.pypi_url = self._url_for_path("pypi") # This is part of a temporary hack used to block installs of PyPI # packages which depend on external urls only necessary until PyPI can # block such packages themselves self.file_storage_domain = file_storage_domain def _url_for_path(self, path: str) -> str: return urllib.parse.urljoin(self.url, path) PyPI = PackageIndex("https://pypi.org/", file_storage_domain="files.pythonhosted.org") TestPyPI = PackageIndex( "https://test.pypi.org/", file_storage_domain="test-files.pythonhosted.org" )	1.01 KB	2026-01-20 12:58
📄 installation_report.py from collections.abc import Sequence from typing import Any from pip._vendor.packaging.markers import default_environment from pip import __version__ from pip._internal.req.req_install import InstallRequirement class InstallationReport: def __init__(self, install_requirements: Sequence[InstallRequirement]): self._install_requirements = install_requirements @classmethod def _install_req_to_dict(cls, ireq: InstallRequirement) -> dict[str, Any]: assert ireq.download_info, f"No download_info for {ireq}" res = { # PEP 610 json for the download URL. download_info.archive_info.hashes may # be absent when the requirement was installed from the wheel cache # and the cache entry was populated by an older pip version that did not # record origin.json. "download_info": ireq.download_info.to_dict(), # is_direct is true if the requirement was a direct URL reference (which # includes editable requirements), and false if the requirement was # downloaded from a PEP 503 index or --find-links. "is_direct": ireq.is_direct, # is_yanked is true if the requirement was yanked from the index, but # was still selected by pip to conform to PEP 592. "is_yanked": ireq.link.is_yanked if ireq.link else False, # requested is true if the requirement was specified by the user (aka # top level requirement), and false if it was installed as a dependency of a # requirement. https://peps.python.org/pep-0376/#requested "requested": ireq.user_supplied, # PEP 566 json encoding for metadata # https://www.python.org/dev/peps/pep-0566/#json-compatible-metadata "metadata": ireq.get_dist().metadata_dict, } if ireq.user_supplied and ireq.extras: # For top level requirements, the list of requested extras, if any. res["requested_extras"] = sorted(ireq.extras) return res def to_dict(self) -> dict[str, Any]: return { "version": "1", "pip_version": __version__, "install": [ self._install_req_to_dict(ireq) for ireq in self._install_requirements ], # https://peps.python.org/pep-0508/#environment-markers # TODO: currently, the resolver uses the default environment to evaluate # environment markers, so that is what we report here. In the future, it # should also take into account options such as --python-version or # --platform, perhaps under the form of an environment_override field? # https://github.com/pypa/pip/issues/11198 "environment": default_environment(), }	2.77 KB	2026-01-20 12:58
📄 link.py from __future__ import annotations import functools import itertools import logging import os import posixpath import re import urllib.parse from collections.abc import Mapping from dataclasses import dataclass from typing import ( TYPE_CHECKING, Any, NamedTuple, ) from pip._internal.utils.deprecation import deprecated from pip._internal.utils.filetypes import WHEEL_EXTENSION from pip._internal.utils.hashes import Hashes from pip._internal.utils.misc import ( pairwise, redact_auth_from_url, split_auth_from_netloc, splitext, ) from pip._internal.utils.urls import path_to_url, url_to_path if TYPE_CHECKING: from pip._internal.index.collector import IndexContent logger = logging.getLogger(__name__) # Order matters, earlier hashes have a precedence over later hashes for what # we will pick to use. _SUPPORTED_HASHES = ("sha512", "sha384", "sha256", "sha224", "sha1", "md5") @dataclass(frozen=True) class LinkHash: """Links to content may have embedded hash values. This class parses those. `name` must be any member of `_SUPPORTED_HASHES`. This class can be converted to and from `ArchiveInfo`. While ArchiveInfo intends to be JSON-serializable to conform to PEP 610, this class contains the logic for parsing a hash name and value for correctness, and then checking whether that hash conforms to a schema with `.is_hash_allowed()`.""" name: str value: str _hash_url_fragment_re = re.compile( # NB: we do not validate that the second group (.) is a valid hex # digest. Instead, we simply keep that string in this class, and then check it # against Hashes when hash-checking is needed. This is easier to debug than # proactively discarding an invalid hex digest, as we handle incorrect hashes # and malformed hashes in the same place. r"[#&]({choices})=([^&])".format( choices="\|".join(re.escape(hash_name) for hash_name in _SUPPORTED_HASHES) ), ) def __post_init__(self) -> None: assert self.name in _SUPPORTED_HASHES @classmethod @functools.cache def find_hash_url_fragment(cls, url: str) -> LinkHash \| None: """Search a string for a checksum algorithm name and encoded output value.""" match = cls._hash_url_fragment_re.search(url) if match is None: return None name, value = match.groups() return cls(name=name, value=value) def as_dict(self) -> dict[str, str]: return {self.name: self.value} def as_hashes(self) -> Hashes: """Return a Hashes instance which checks only for the current hash.""" return Hashes({self.name: [self.value]}) def is_hash_allowed(self, hashes: Hashes \| None) -> bool: """ Return True if the current hash is allowed by `hashes`. """ if hashes is None: return False return hashes.is_hash_allowed(self.name, hex_digest=self.value) @dataclass(frozen=True) class MetadataFile: """Information about a core metadata file associated with a distribution.""" hashes: dict[str, str] \| None def __post_init__(self) -> None: if self.hashes is not None: assert all(name in _SUPPORTED_HASHES for name in self.hashes) def supported_hashes(hashes: dict[str, str] \| None) -> dict[str, str] \| None: # Remove any unsupported hash types from the mapping. If this leaves no # supported hashes, return None if hashes is None: return None hashes = {n: v for n, v in hashes.items() if n in _SUPPORTED_HASHES} if not hashes: return None return hashes def _clean_url_path_part(part: str) -> str: """ Clean a "part" of a URL path (i.e. after splitting on "@" characters). """ # We unquote prior to quoting to make sure nothing is double quoted. return urllib.parse.quote(urllib.parse.unquote(part)) def _clean_file_url_path(part: str) -> str: """ Clean the first part of a URL path that corresponds to a local filesystem path (i.e. the first part after splitting on "@" characters). """ # We unquote prior to quoting to make sure nothing is double quoted. # Also, on Windows the path part might contain a drive letter which # should not be quoted. On Linux where drive letters do not # exist, the colon should be quoted. We rely on urllib.request # to do the right thing here. ret = urllib.request.pathname2url(urllib.request.url2pathname(part)) if ret.startswith("///"): # Remove any URL authority section, leaving only the URL path. ret = ret.removeprefix("//") return ret # percent-encoded: / _reserved_chars_re = re.compile("(@\|%2F)", re.IGNORECASE) def _clean_url_path(path: str, is_local_path: bool) -> str: """ Clean the path portion of a URL. """ if is_local_path: clean_func = _clean_file_url_path else: clean_func = _clean_url_path_part # Split on the reserved characters prior to cleaning so that # revision strings in VCS URLs are properly preserved. parts = _reserved_chars_re.split(path) cleaned_parts = [] for to_clean, reserved in pairwise(itertools.chain(parts, [""])): cleaned_parts.append(clean_func(to_clean)) # Normalize %xx escapes (e.g. %2f -> %2F) cleaned_parts.append(reserved.upper()) return "".join(cleaned_parts) def _ensure_quoted_url(url: str) -> str: """ Make sure a link is fully quoted. For example, if ' ' occurs in the URL, it will be replaced with "%20", and without double-quoting other characters. """ # Split the URL into parts according to the general structure # `scheme://netloc/path?query#fragment`. result = urllib.parse.urlsplit(url) # If the netloc is empty, then the URL refers to a local filesystem path. is_local_path = not result.netloc path = _clean_url_path(result.path, is_local_path=is_local_path) # Temporarily replace scheme with file to ensure the URL generated by # urlunsplit() contains an empty netloc (file://) as per RFC 1738. ret = urllib.parse.urlunsplit(result._replace(scheme="file", path=path)) ret = result.scheme + ret[4:] # Restore original scheme. return ret def _absolute_link_url(base_url: str, url: str) -> str: """ A faster implementation of urllib.parse.urljoin with a shortcut for absolute http/https URLs. """ if url.startswith(("https://", "http://")): return url else: return urllib.parse.urljoin(base_url, url) @functools.total_ordering class Link: """Represents a parsed link from a Package Index's simple URL""" __slots__ = [ "_parsed_url", "_url", "_path", "_hashes", "comes_from", "requires_python", "yanked_reason", "metadata_file_data", "cache_link_parsing", "egg_fragment", ] def __init__( self, url: str, comes_from: str \| IndexContent \| None = None, requires_python: str \| None = None, yanked_reason: str \| None = None, metadata_file_data: MetadataFile \| None = None, cache_link_parsing: bool = True, hashes: Mapping[str, str] \| None = None, ) -> None: """ :param url: url of the resource pointed to (href of the link) :param comes_from: instance of IndexContent where the link was found, or string. :param requires_python: String containing the `Requires-Python` metadata field, specified in PEP 345. This may be specified by a data-requires-python attribute in the HTML link tag, as described in PEP 503. :param yanked_reason: the reason the file has been yanked, if the file has been yanked, or None if the file hasn't been yanked. This is the value of the "data-yanked" attribute, if present, in a simple repository HTML link. If the file has been yanked but no reason was provided, this should be the empty string. See PEP 592 for more information and the specification. :param metadata_file_data: the metadata attached to the file, or None if no such metadata is provided. This argument, if not None, indicates that a separate metadata file exists, and also optionally supplies hashes for that file. :param cache_link_parsing: A flag that is used elsewhere to determine whether resources retrieved from this link should be cached. PyPI URLs should generally have this set to False, for example. :param hashes: A mapping of hash names to digests to allow us to determine the validity of a download. """ # The comes_from, requires_python, and metadata_file_data arguments are # only used by classmethods of this class, and are not used in client # code directly. # url can be a UNC windows share if url.startswith("\\\\"): url = path_to_url(url) self._parsed_url = urllib.parse.urlsplit(url) # Store the url as a private attribute to prevent accidentally # trying to set a new value. self._url = url # The .path property is hot, so calculate its value ahead of time. self._path = urllib.parse.unquote(self._parsed_url.path) link_hash = LinkHash.find_hash_url_fragment(url) hashes_from_link = {} if link_hash is None else link_hash.as_dict() if hashes is None: self._hashes = hashes_from_link else: self._hashes = {hashes, hashes_from_link} self.comes_from = comes_from self.requires_python = requires_python if requires_python else None self.yanked_reason = yanked_reason self.metadata_file_data = metadata_file_data self.cache_link_parsing = cache_link_parsing self.egg_fragment = self._egg_fragment() @classmethod def from_json( cls, file_data: dict[str, Any], page_url: str, ) -> Link \| None: """ Convert an pypi json document from a simple repository page into a Link. """ file_url = file_data.get("url") if file_url is None: return None url = _ensure_quoted_url(_absolute_link_url(page_url, file_url)) pyrequire = file_data.get("requires-python") yanked_reason = file_data.get("yanked") hashes = file_data.get("hashes", {}) # PEP 714: Indexes must use the name core-metadata, but # clients should support the old name as a fallback for compatibility. metadata_info = file_data.get("core-metadata") if metadata_info is None: metadata_info = file_data.get("dist-info-metadata") # The metadata info value may be a boolean, or a dict of hashes. if isinstance(metadata_info, dict): # The file exists, and hashes have been supplied metadata_file_data = MetadataFile(supported_hashes(metadata_info)) elif metadata_info: # The file exists, but there are no hashes metadata_file_data = MetadataFile(None) else: # False or not present: the file does not exist metadata_file_data = None # The Link.yanked_reason expects an empty string instead of a boolean. if yanked_reason and not isinstance(yanked_reason, str): yanked_reason = "" # The Link.yanked_reason expects None instead of False. elif not yanked_reason: yanked_reason = None return cls( url, comes_from=page_url, requires_python=pyrequire, yanked_reason=yanked_reason, hashes=hashes, metadata_file_data=metadata_file_data, ) @classmethod def from_element( cls, anchor_attribs: dict[str, str \| None], page_url: str, base_url: str, ) -> Link \| None: """ Convert an anchor element's attributes in a simple repository page to a Link. """ href = anchor_attribs.get("href") if not href: return None url = _ensure_quoted_url(_absolute_link_url(base_url, href)) pyrequire = anchor_attribs.get("data-requires-python") yanked_reason = anchor_attribs.get("data-yanked") # PEP 714: Indexes must use the name data-core-metadata, but # clients should support the old name as a fallback for compatibility. metadata_info = anchor_attribs.get("data-core-metadata") if metadata_info is None: metadata_info = anchor_attribs.get("data-dist-info-metadata") # The metadata info value may be the string "true", or a string of # the form "hashname=hashval" if metadata_info == "true": # The file exists, but there are no hashes metadata_file_data = MetadataFile(None) elif metadata_info is None: # The file does not exist metadata_file_data = None else: # The file exists, and hashes have been supplied hashname, sep, hashval = metadata_info.partition("=") if sep == "=": metadata_file_data = MetadataFile(supported_hashes({hashname: hashval})) else: # Error - data is wrong. Treat as no hashes supplied. logger.debug( "Index returned invalid data-dist-info-metadata value: %s", metadata_info, ) metadata_file_data = MetadataFile(None) return cls( url, comes_from=page_url, requires_python=pyrequire, yanked_reason=yanked_reason, metadata_file_data=metadata_file_data, ) def __str__(self) -> str: if self.requires_python: rp = f" (requires-python:{self.requires_python})" else: rp = "" if self.comes_from: return f"{self.redacted_url} (from {self.comes_from}){rp}" else: return self.redacted_url def __repr__(self) -> str: return f"<Link {self}>" def __hash__(self) -> int: return hash(self.url) def __eq__(self, other: Any) -> bool: if not isinstance(other, Link): return NotImplemented return self.url == other.url def __lt__(self, other: Any) -> bool: if not isinstance(other, Link): return NotImplemented return self.url < other.url @property def url(self) -> str: return self._url @property def redacted_url(self) -> str: return redact_auth_from_url(self.url) @property def filename(self) -> str: path = self.path.rstrip("/") name = posixpath.basename(path) if not name: # Make sure we don't leak auth information if the netloc # includes a username and password. netloc, user_pass = split_auth_from_netloc(self.netloc) return netloc name = urllib.parse.unquote(name) assert name, f"URL {self._url!r} produced no filename" return name @property def file_path(self) -> str: return url_to_path(self.url) @property def scheme(self) -> str: return self._parsed_url.scheme @property def netloc(self) -> str: """ This can contain auth information. """ return self._parsed_url.netloc @property def path(self) -> str: return self._path def splitext(self) -> tuple[str, str]: return splitext(posixpath.basename(self.path.rstrip("/"))) @property def ext(self) -> str: return self.splitext()[1] @property def url_without_fragment(self) -> str: scheme, netloc, path, query, fragment = self._parsed_url return urllib.parse.urlunsplit((scheme, netloc, path, query, "")) _egg_fragment_re = re.compile(r"[#&]egg=([^&])") # Per PEP 508. _project_name_re = re.compile( r"^([A-Z0-9]\|[A-Z0-9][A-Z0-9._-][A-Z0-9])$", re.IGNORECASE ) def _egg_fragment(self) -> str \| None: match = self._egg_fragment_re.search(self._url) if not match: return None # An egg fragment looks like a PEP 508 project name, along with # an optional extras specifier. Anything else is invalid. project_name = match.group(1) if not self._project_name_re.match(project_name): deprecated( reason=f"{self} contains an egg fragment with a non-PEP 508 name.", replacement="to use the req @ url syntax, and remove the egg fragment", gone_in="26.0", issue=13157, ) return project_name _subdirectory_fragment_re = re.compile(r"[#&]subdirectory=([^&]*)") @property def subdirectory_fragment(self) -> str \| None: match = self._subdirectory_fragment_re.search(self._url) if not match: return None return match.group(1) def metadata_link(self) -> Link \| None: """Return a link to the associated core metadata file (if any).""" if self.metadata_file_data is None: return None metadata_url = f"{self.url_without_fragment}.metadata" if self.metadata_file_data.hashes is None: return Link(metadata_url) return Link(metadata_url, hashes=self.metadata_file_data.hashes) def as_hashes(self) -> Hashes: return Hashes({k: [v] for k, v in self._hashes.items()}) @property def hash(self) -> str \| None: return next(iter(self._hashes.values()), None) @property def hash_name(self) -> str \| None: return next(iter(self._hashes), None) @property def show_url(self) -> str: return posixpath.basename(self._url.split("#", 1)[0].split("?", 1)[0]) @property def is_file(self) -> bool: return self.scheme == "file" def is_existing_dir(self) -> bool: return self.is_file and os.path.isdir(self.file_path) @property def is_wheel(self) -> bool: return self.ext == WHEEL_EXTENSION @property def is_vcs(self) -> bool: from pip._internal.vcs import vcs return self.scheme in vcs.all_schemes @property def is_yanked(self) -> bool: return self.yanked_reason is not None @property def has_hash(self) -> bool: return bool(self._hashes) def is_hash_allowed(self, hashes: Hashes \| None) -> bool: """ Return True if the link has a hash and it is allowed by `hashes`. """ if hashes is None: return False return any(hashes.is_hash_allowed(k, v) for k, v in self._hashes.items()) class _CleanResult(NamedTuple): """Convert link for equivalency check. This is used in the resolver to check whether two URL-specified requirements likely point to the same distribution and can be considered equivalent. This equivalency logic avoids comparing URLs literally, which can be too strict (e.g. "a=1&b=2" vs "b=2&a=1") and produce conflicts unexpecting to users. Currently this does three things: 1. Drop the basic auth part. This is technically wrong since a server can serve different content based on auth, but if it does that, it is even impossible to guarantee two URLs without auth are equivalent, since the user can input different auth information when prompted. So the practical solution is to assume the auth doesn't affect the response. 2. Parse the query to avoid the ordering issue. Note that ordering under the same key in the query are NOT cleaned; i.e. "a=1&a=2" and "a=2&a=1" are still considered different. 3. Explicitly drop most of the fragment part, except ``subdirectory=`` and hash values, since it should have no impact the downloaded content. Note that this drops the "egg=" part historically used to denote the requested project (and extras), which is wrong in the strictest sense, but too many people are supplying it inconsistently to cause superfluous resolution conflicts, so we choose to also ignore them. """ parsed: urllib.parse.SplitResult query: dict[str, list[str]] subdirectory: str hashes: dict[str, str] def _clean_link(link: Link) -> _CleanResult: parsed = link._parsed_url netloc = parsed.netloc.rsplit("@", 1)[-1] # According to RFC 8089, an empty host in file: means localhost. if parsed.scheme == "file" and not netloc: netloc = "localhost" fragment = urllib.parse.parse_qs(parsed.fragment) if "egg" in fragment: logger.debug("Ignoring egg= fragment in %s", link) try: # If there are multiple subdirectory values, use the first one. # This matches the behavior of Link.subdirectory_fragment. subdirectory = fragment["subdirectory"][0] except (IndexError, KeyError): subdirectory = "" # If there are multiple hash values under the same algorithm, use the # first one. This matches the behavior of Link.hash_value. hashes = {k: fragment[k][0] for k in _SUPPORTED_HASHES if k in fragment} return _CleanResult( parsed=parsed._replace(netloc=netloc, query="", fragment=""), query=urllib.parse.parse_qs(parsed.query), subdirectory=subdirectory, hashes=hashes, ) @functools.cache def links_equivalent(link1: Link, link2: Link) -> bool: return _clean_link(link1) == _clean_link(link2)	21.28 KB	2026-01-20 12:58
📄 pylock.py from __future__ import annotations import dataclasses import re from collections.abc import Iterable from dataclasses import dataclass from pathlib import Path from typing import TYPE_CHECKING, Any from pip._vendor import tomli_w from pip._internal.models.direct_url import ArchiveInfo, DirInfo, VcsInfo from pip._internal.models.link import Link from pip._internal.req.req_install import InstallRequirement from pip._internal.utils.urls import url_to_path if TYPE_CHECKING: from typing_extensions import Self PYLOCK_FILE_NAME_RE = re.compile(r"^pylock\.([^.]+)\.toml$") def is_valid_pylock_file_name(path: Path) -> bool: return path.name == "pylock.toml" or bool(re.match(PYLOCK_FILE_NAME_RE, path.name)) def _toml_dict_factory(data: list[tuple[str, Any]]) -> dict[str, Any]: return {key.replace("_", "-"): value for key, value in data if value is not None} @dataclass class PackageVcs: type: str url: str \| None # (not supported) path: Optional[str] requested_revision: str \| None commit_id: str subdirectory: str \| None @dataclass class PackageDirectory: path: str editable: bool \| None subdirectory: str \| None @dataclass class PackageArchive: url: str \| None # (not supported) path: Optional[str] # (not supported) size: Optional[int] # (not supported) upload_time: Optional[datetime] hashes: dict[str, str] subdirectory: str \| None @dataclass class PackageSdist: name: str # (not supported) upload_time: Optional[datetime] url: str \| None # (not supported) path: Optional[str] # (not supported) size: Optional[int] hashes: dict[str, str] @dataclass class PackageWheel: name: str # (not supported) upload_time: Optional[datetime] url: str \| None # (not supported) path: Optional[str] # (not supported) size: Optional[int] hashes: dict[str, str] @dataclass class Package: name: str version: str \| None = None # (not supported) marker: Optional[str] # (not supported) requires_python: Optional[str] # (not supported) dependencies vcs: PackageVcs \| None = None directory: PackageDirectory \| None = None archive: PackageArchive \| None = None # (not supported) index: Optional[str] sdist: PackageSdist \| None = None wheels: list[PackageWheel] \| None = None # (not supported) attestation_identities: Optional[List[Dict[str, Any]]] # (not supported) tool: Optional[Dict[str, Any]] @classmethod def from_install_requirement(cls, ireq: InstallRequirement, base_dir: Path) -> Self: base_dir = base_dir.resolve() dist = ireq.get_dist() download_info = ireq.download_info assert download_info package = cls(name=dist.canonical_name) if ireq.is_direct: if isinstance(download_info.info, VcsInfo): package.vcs = PackageVcs( type=download_info.info.vcs, url=download_info.url, requested_revision=download_info.info.requested_revision, commit_id=download_info.info.commit_id, subdirectory=download_info.subdirectory, ) elif isinstance(download_info.info, DirInfo): package.directory = PackageDirectory( path=( Path(url_to_path(download_info.url)) .resolve() .relative_to(base_dir) .as_posix() ), editable=( download_info.info.editable if download_info.info.editable else None ), subdirectory=download_info.subdirectory, ) elif isinstance(download_info.info, ArchiveInfo): if not download_info.info.hashes: raise NotImplementedError() package.archive = PackageArchive( url=download_info.url, hashes=download_info.info.hashes, subdirectory=download_info.subdirectory, ) else: # should never happen raise NotImplementedError() else: package.version = str(dist.version) if isinstance(download_info.info, ArchiveInfo): if not download_info.info.hashes: raise NotImplementedError() link = Link(download_info.url) if link.is_wheel: package.wheels = [ PackageWheel( name=link.filename, url=download_info.url, hashes=download_info.info.hashes, ) ] else: package.sdist = PackageSdist( name=link.filename, url=download_info.url, hashes=download_info.info.hashes, ) else: # should never happen raise NotImplementedError() return package @dataclass class Pylock: lock_version: str = "1.0" # (not supported) environments: Optional[List[str]] # (not supported) requires_python: Optional[str] # (not supported) extras: List[str] = [] # (not supported) dependency_groups: List[str] = [] created_by: str = "pip" packages: list[Package] = dataclasses.field(default_factory=list) # (not supported) tool: Optional[Dict[str, Any]] def as_toml(self) -> str: return tomli_w.dumps(dataclasses.asdict(self, dict_factory=_toml_dict_factory)) @classmethod def from_install_requirements( cls, install_requirements: Iterable[InstallRequirement], base_dir: Path ) -> Self: return cls( packages=sorted( ( Package.from_install_requirement(ireq, base_dir) for ireq in install_requirements ), key=lambda p: p.name, ) )	6.07 KB	2026-01-20 12:58
📄 scheme.py """ For types associated with installation schemes. For a general overview of available schemes and their context, see https://docs.python.org/3/install/index.html#alternate-installation. """ from dataclasses import dataclass SCHEME_KEYS = ["platlib", "purelib", "headers", "scripts", "data"] @dataclass(frozen=True) class Scheme: """A Scheme holds paths which are used as the base directories for artifacts associated with a Python package. """ __slots__ = SCHEME_KEYS platlib: str purelib: str headers: str scripts: str data: str	575 B	2026-01-20 12:58
📄 search_scope.py import itertools import logging import os import posixpath import urllib.parse from dataclasses import dataclass from pip._vendor.packaging.utils import canonicalize_name from pip._internal.models.index import PyPI from pip._internal.utils.compat import has_tls from pip._internal.utils.misc import normalize_path, redact_auth_from_url logger = logging.getLogger(__name__) @dataclass(frozen=True) class SearchScope: """ Encapsulates the locations that pip is configured to search. """ __slots__ = ["find_links", "index_urls", "no_index"] find_links: list[str] index_urls: list[str] no_index: bool @classmethod def create( cls, find_links: list[str], index_urls: list[str], no_index: bool, ) -> "SearchScope": """ Create a SearchScope object after normalizing the `find_links`. """ # Build find_links. If an argument starts with ~, it may be # a local file relative to a home directory. So try normalizing # it and if it exists, use the normalized version. # This is deliberately conservative - it might be fine just to # blindly normalize anything starting with a ~... built_find_links: list[str] = [] for link in find_links: if link.startswith("~"): new_link = normalize_path(link) if os.path.exists(new_link): link = new_link built_find_links.append(link) # If we don't have TLS enabled, then WARN if anyplace we're looking # relies on TLS. if not has_tls(): for link in itertools.chain(index_urls, built_find_links): parsed = urllib.parse.urlparse(link) if parsed.scheme == "https": logger.warning( "pip is configured with locations that require " "TLS/SSL, however the ssl module in Python is not " "available." ) break return cls( find_links=built_find_links, index_urls=index_urls, no_index=no_index, ) def get_formatted_locations(self) -> str: lines = [] redacted_index_urls = [] if self.index_urls and self.index_urls != [PyPI.simple_url]: for url in self.index_urls: redacted_index_url = redact_auth_from_url(url) # Parse the URL purl = urllib.parse.urlsplit(redacted_index_url) # URL is generally invalid if scheme and netloc is missing # there are issues with Python and URL parsing, so this test # is a bit crude. See bpo-20271, bpo-23505. Python doesn't # always parse invalid URLs correctly - it should raise # exceptions for malformed URLs if not purl.scheme and not purl.netloc: logger.warning( 'The index url "%s" seems invalid, please provide a scheme.', redacted_index_url, ) redacted_index_urls.append(redacted_index_url) lines.append( "Looking in indexes: {}".format(", ".join(redacted_index_urls)) ) if self.find_links: lines.append( "Looking in links: {}".format( ", ".join(redact_auth_from_url(url) for url in self.find_links) ) ) return "\n".join(lines) def get_index_urls_locations(self, project_name: str) -> list[str]: """Returns the locations found via self.index_urls Checks the url_name on the main (first in the list) index and use this url_name to produce all locations """ def mkurl_pypi_url(url: str) -> str: loc = posixpath.join( url, urllib.parse.quote(canonicalize_name(project_name)) ) # For maximum compatibility with easy_install, ensure the path # ends in a trailing slash. Although this isn't in the spec # (and PyPI can handle it without the slash) some other index # implementations might break if they relied on easy_install's # behavior. if not loc.endswith("/"): loc = loc + "/" return loc return [mkurl_pypi_url(url) for url in self.index_urls]	4.4 KB	2026-01-20 12:58
📄 selection_prefs.py from __future__ import annotations from pip._internal.models.format_control import FormatControl # TODO: This needs Python 3.10's improved slots support for dataclasses # to be converted into a dataclass. class SelectionPreferences: """ Encapsulates the candidate selection preferences for downloading and installing files. """ __slots__ = [ "allow_yanked", "allow_all_prereleases", "format_control", "prefer_binary", "ignore_requires_python", ] # Don't include an allow_yanked default value to make sure each call # site considers whether yanked releases are allowed. This also causes # that decision to be made explicit in the calling code, which helps # people when reading the code. def __init__( self, allow_yanked: bool, allow_all_prereleases: bool = False, format_control: FormatControl \| None = None, prefer_binary: bool = False, ignore_requires_python: bool \| None = None, ) -> None: """Create a SelectionPreferences object. :param allow_yanked: Whether files marked as yanked (in the sense of PEP 592) are permitted to be candidates for install. :param format_control: A FormatControl object or None. Used to control the selection of source packages / binary packages when consulting the index and links. :param prefer_binary: Whether to prefer an old, but valid, binary dist over a new source dist. :param ignore_requires_python: Whether to ignore incompatible "Requires-Python" values in links. Defaults to False. """ if ignore_requires_python is None: ignore_requires_python = False self.allow_yanked = allow_yanked self.allow_all_prereleases = allow_all_prereleases self.format_control = format_control self.prefer_binary = prefer_binary self.ignore_requires_python = ignore_requires_python	1.97 KB	2026-01-20 12:58
📄 target_python.py from __future__ import annotations import sys from pip._vendor.packaging.tags import Tag from pip._internal.utils.compatibility_tags import get_supported, version_info_to_nodot from pip._internal.utils.misc import normalize_version_info class TargetPython: """ Encapsulates the properties of a Python interpreter one is targeting for a package install, download, etc. """ __slots__ = [ "_given_py_version_info", "abis", "implementation", "platforms", "py_version", "py_version_info", "_valid_tags", "_valid_tags_set", ] def __init__( self, platforms: list[str] \| None = None, py_version_info: tuple[int, ...] \| None = None, abis: list[str] \| None = None, implementation: str \| None = None, ) -> None: """ :param platforms: A list of strings or None. If None, searches for packages that are supported by the current system. Otherwise, will find packages that can be built on the platforms passed in. These packages will only be downloaded for distribution: they will not be built locally. :param py_version_info: An optional tuple of ints representing the Python version information to use (e.g. `sys.version_info[:3]`). This can have length 1, 2, or 3 when provided. :param abis: A list of strings or None. This is passed to compatibility_tags.py's get_supported() function as is. :param implementation: A string or None. This is passed to compatibility_tags.py's get_supported() function as is. """ # Store the given py_version_info for when we call get_supported(). self._given_py_version_info = py_version_info if py_version_info is None: py_version_info = sys.version_info[:3] else: py_version_info = normalize_version_info(py_version_info) py_version = ".".join(map(str, py_version_info[:2])) self.abis = abis self.implementation = implementation self.platforms = platforms self.py_version = py_version self.py_version_info = py_version_info # This is used to cache the return value of get_(un)sorted_tags. self._valid_tags: list[Tag] \| None = None self._valid_tags_set: set[Tag] \| None = None def format_given(self) -> str: """ Format the given, non-None attributes for display. """ display_version = None if self._given_py_version_info is not None: display_version = ".".join( str(part) for part in self._given_py_version_info ) key_values = [ ("platforms", self.platforms), ("version_info", display_version), ("abis", self.abis), ("implementation", self.implementation), ] return " ".join( f"{key}={value!r}" for key, value in key_values if value is not None ) def get_sorted_tags(self) -> list[Tag]: """ Return the supported PEP 425 tags to check wheel candidates against. The tags are returned in order of preference (most preferred first). """ if self._valid_tags is None: # Pass versions=None if no py_version_info was given since # versions=None uses special default logic. py_version_info = self._given_py_version_info if py_version_info is None: version = None else: version = version_info_to_nodot(py_version_info) tags = get_supported( version=version, platforms=self.platforms, abis=self.abis, impl=self.implementation, ) self._valid_tags = tags return self._valid_tags def get_unsorted_tags(self) -> set[Tag]: """Exactly the same as get_sorted_tags, but returns a set. This is important for performance. """ if self._valid_tags_set is None: self._valid_tags_set = set(self.get_sorted_tags()) return self._valid_tags_set	4.14 KB	2026-01-20 12:58
📄 wheel.py """Represents a wheel file and provides access to the various parts of the name that have meaning. """ from __future__ import annotations from collections.abc import Iterable from pip._vendor.packaging.tags import Tag from pip._vendor.packaging.utils import ( InvalidWheelFilename as _PackagingInvalidWheelFilename, ) from pip._vendor.packaging.utils import parse_wheel_filename from pip._internal.exceptions import InvalidWheelFilename class Wheel: """A wheel file""" def __init__(self, filename: str) -> None: self.filename = filename try: wheel_info = parse_wheel_filename(filename) except _PackagingInvalidWheelFilename as e: raise InvalidWheelFilename(e.args[0]) from None self.name, _version, self.build_tag, self.file_tags = wheel_info self.version = str(_version) def get_formatted_file_tags(self) -> list[str]: """Return the wheel's tags as a sorted list of strings.""" return sorted(str(tag) for tag in self.file_tags) def support_index_min(self, tags: list[Tag]) -> int: """Return the lowest index that one of the wheel's file_tag combinations achieves in the given list of supported tags. For example, if there are 8 supported tags and one of the file tags is first in the list, then return 0. :param tags: the PEP 425 tags to check the wheel against, in order with most preferred first. :raises ValueError: If none of the wheel's file tags match one of the supported tags. """ try: return next(i for i, t in enumerate(tags) if t in self.file_tags) except StopIteration: raise ValueError() def find_most_preferred_tag( self, tags: list[Tag], tag_to_priority: dict[Tag, int] ) -> int: """Return the priority of the most preferred tag that one of the wheel's file tag combinations achieves in the given list of supported tags using the given tag_to_priority mapping, where lower priorities are more-preferred. This is used in place of support_index_min in some cases in order to avoid an expensive linear scan of a large list of tags. :param tags: the PEP 425 tags to check the wheel against. :param tag_to_priority: a mapping from tag to priority of that tag, where lower is more preferred. :raises ValueError: If none of the wheel's file tags match one of the supported tags. """ return min( tag_to_priority[tag] for tag in self.file_tags if tag in tag_to_priority ) def supported(self, tags: Iterable[Tag]) -> bool: """Return whether the wheel is compatible with one of the given tags. :param tags: the PEP 425 tags to check the wheel against. """ return not self.file_tags.isdisjoint(tags)	2.85 KB	2026-01-20 12:58

Terminal