Загрузить файлы в «venv/Lib/site-packages/pip/_internal/req»

venv/Lib/site-packages/pip/_internal/req/__init__.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import collections
+import logging
+from collections.abc import Generator
+from dataclasses import dataclass
+
+from pip._internal.cli.progress_bars import BarType, get_install_progress_renderer
+from pip._internal.utils.logging import indent_log
+
+from .req_file import parse_requirements
+from .req_install import InstallRequirement
+from .req_set import RequirementSet
+
+__all__ = [
+    "RequirementSet",
+    "InstallRequirement",
+    "parse_requirements",
+    "install_given_reqs",
+]
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class InstallationResult:
+    name: str
+
+
+def _validate_requirements(
+    requirements: list[InstallRequirement],
+) -> Generator[tuple[str, InstallRequirement], None, None]:
+    for req in requirements:
+        assert req.name, f"invalid to-be-installed requirement: {req}"
+        yield req.name, req
+
+
+def install_given_reqs(
+    requirements: list[InstallRequirement],
+    root: str | None,
+    home: str | None,
+    prefix: str | None,
+    warn_script_location: bool,
+    use_user_site: bool,
+    pycompile: bool,
+    progress_bar: BarType,
+) -> list[InstallationResult]:
+    """
+    Install everything in the given list.
+
+    (to be called after having downloaded and unpacked the packages)
+    """
+    to_install = collections.OrderedDict(_validate_requirements(requirements))
+
+    if to_install:
+        logger.info(
+            "Installing collected packages: %s",
+            ", ".join(to_install.keys()),
+        )
+
+    installed = []
+
+    show_progress = logger.isEnabledFor(logging.INFO) and len(to_install) > 1
+
+    items = iter(to_install.values())
+    if show_progress:
+        renderer = get_install_progress_renderer(
+            bar_type=progress_bar, total=len(to_install)
+        )
+        items = renderer(items)
+
+    with indent_log():
+        for requirement in items:
+            req_name = requirement.name
+            assert req_name is not None
+            if requirement.should_reinstall:
+                logger.info("Attempting uninstall: %s", req_name)
+                with indent_log():
+                    uninstalled_pathset = requirement.uninstall(auto_confirm=True)
+            else:
+                uninstalled_pathset = None
+
+            try:
+                requirement.install(
+                    root=root,
+                    home=home,
+                    prefix=prefix,
+                    warn_script_location=warn_script_location,
+                    use_user_site=use_user_site,
+                    pycompile=pycompile,
+                )
+            except Exception:
+                # if install did not succeed, rollback previous uninstall
+                if uninstalled_pathset and not requirement.install_succeeded:
+                    uninstalled_pathset.rollback()
+                raise
+            else:
+                if uninstalled_pathset and requirement.install_succeeded:
+                    uninstalled_pathset.commit()
+
+            installed.append(InstallationResult(req_name))
+
+    return installed

venv/Lib/site-packages/pip/_internal/req/constructors.py

@@ -0,0 +1,568 @@
+"""Backing implementation for InstallRequirement's various constructors
+
+The idea here is that these formed a major chunk of InstallRequirement's size
+so, moving them and support code dedicated to them outside of that class
+helps creates for better understandability for the rest of the code.
+
+These are meant to be used elsewhere within pip to create instances of
+InstallRequirement.
+"""
+
+from __future__ import annotations
+
+import copy
+import logging
+import os
+import re
+from collections.abc import Collection
+from dataclasses import dataclass
+
+from pip._vendor.packaging.markers import Marker
+from pip._vendor.packaging.requirements import InvalidRequirement, Requirement
+from pip._vendor.packaging.specifiers import Specifier
+
+from pip._internal.exceptions import InstallationError
+from pip._internal.models.index import PyPI, TestPyPI
+from pip._internal.models.link import Link
+from pip._internal.models.wheel import Wheel
+from pip._internal.req.req_file import ParsedRequirement
+from pip._internal.req.req_install import InstallRequirement
+from pip._internal.utils.filetypes import is_archive_file
+from pip._internal.utils.misc import is_installable_dir
+from pip._internal.utils.packaging import get_requirement
+from pip._internal.utils.urls import path_to_url
+from pip._internal.vcs import is_url, vcs
+
+__all__ = [
+    "install_req_from_editable",
+    "install_req_from_line",
+    "parse_editable",
+]
+
+logger = logging.getLogger(__name__)
+operators = Specifier._operators.keys()
+
+
+def _strip_extras(path: str) -> tuple[str, str | None]:
+    m = re.match(r"^(.+)(\[[^\]]+\])$", path)
+    extras = None
+    if m:
+        path_no_extras = m.group(1).rstrip()
+        extras = m.group(2)
+    else:
+        path_no_extras = path
+
+    return path_no_extras, extras
+
+
+def convert_extras(extras: str | None) -> set[str]:
+    if not extras:
+        return set()
+    return get_requirement("placeholder" + extras.lower()).extras
+
+
+def _set_requirement_extras(req: Requirement, new_extras: set[str]) -> Requirement:
+    """
+    Returns a new requirement based on the given one, with the supplied extras. If the
+    given requirement already has extras those are replaced (or dropped if no new extras
+    are given).
+    """
+    match: re.Match[str] | None = re.fullmatch(
+        # see https://peps.python.org/pep-0508/#complete-grammar
+        r"([\w\t .-]+)(\[[^\]]*\])?(.*)",
+        str(req),
+        flags=re.ASCII,
+    )
+    # ireq.req is a valid requirement so the regex should always match
+    assert (
+        match is not None
+    ), f"regex match on requirement {req} failed, this should never happen"
+    pre: str | None = match.group(1)
+    post: str | None = match.group(3)
+    assert (
+        pre is not None and post is not None
+    ), f"regex group selection for requirement {req} failed, this should never happen"
+    extras: str = "[{}]".format(",".join(sorted(new_extras)) if new_extras else "")
+    return get_requirement(f"{pre}{extras}{post}")
+
+
+def _parse_direct_url_editable(editable_req: str) -> tuple[str | None, str, set[str]]:
+    try:
+        req = Requirement(editable_req)
+    except InvalidRequirement:
+        pass
+    else:
+        if req.url:
+            # Join the marker back into the name part. This will be parsed out
+            # later into a Requirement again.
+            if req.marker:
+                name = f"{req.name} ; {req.marker}"
+            else:
+                name = req.name
+            return (name, req.url, req.extras)
+
+    raise ValueError
+
+
+def _parse_pip_syntax_editable(editable_req: str) -> tuple[str | None, str, set[str]]:
+    url = editable_req
+
+    # If a file path is specified with extras, strip off the extras.
+    url_no_extras, extras = _strip_extras(url)
+
+    if os.path.isdir(url_no_extras):
+        # Treating it as code that has already been checked out
+        url_no_extras = path_to_url(url_no_extras)
+
+    if url_no_extras.lower().startswith("file:"):
+        package_name = Link(url_no_extras).egg_fragment
+        if extras:
+            return (
+                package_name,
+                url_no_extras,
+                get_requirement("placeholder" + extras.lower()).extras,
+            )
+        else:
+            return package_name, url_no_extras, set()
+
+    for version_control in vcs:
+        if url.lower().startswith(f"{version_control}:"):
+            url = f"{version_control}+{url}"
+            break
+
+    return Link(url).egg_fragment, url, set()
+
+
+def parse_editable(editable_req: str) -> tuple[str | None, str, set[str]]:
+    """Parses an editable requirement into:
+        - a requirement name with environment markers
+        - an URL
+        - extras
+    Accepted requirements:
+        - svn+http://blahblah@rev#egg=Foobar[baz]&subdirectory=version_subdir
+        - local_path[some_extra]
+        - Foobar[extra] @ svn+http://blahblah@rev#subdirectory=subdir ; markers
+    """
+    try:
+        package_name, url, extras = _parse_direct_url_editable(editable_req)
+    except ValueError:
+        package_name, url, extras = _parse_pip_syntax_editable(editable_req)
+
+    link = Link(url)
+
+    if not link.is_vcs and not link.url.startswith("file:"):
+        backends = ", ".join(vcs.all_schemes)
+        raise InstallationError(
+            f"{editable_req} is not a valid editable requirement. "
+            f"It should either be a path to a local project or a VCS URL "
+            f"(beginning with {backends})."
+        )
+
+    # The project name can be inferred from local file URIs easily.
+    if not package_name and not link.url.startswith("file:"):
+        raise InstallationError(
+            f"Could not detect requirement name for '{editable_req}', "
+            "please specify one with your_package_name @ URL"
+        )
+    return package_name, url, extras
+
+
+def check_first_requirement_in_file(filename: str) -> None:
+    """Check if file is parsable as a requirements file.
+
+    This is heavily based on ``pkg_resources.parse_requirements``, but
+    simplified to just check the first meaningful line.
+
+    :raises InvalidRequirement: If the first meaningful line cannot be parsed
+        as an requirement.
+    """
+    with open(filename, encoding="utf-8", errors="ignore") as f:
+        # Create a steppable iterator, so we can handle \-continuations.
+        lines = (
+            line
+            for line in (line.strip() for line in f)
+            if line and not line.startswith("#")  # Skip blank lines/comments.
+        )
+
+        for line in lines:
+            # Drop comments -- a hash without a space may be in a URL.
+            if " #" in line:
+                line = line[: line.find(" #")]
+            # If there is a line continuation, drop it, and append the next line.
+            if line.endswith("\\"):
+                line = line[:-2].strip() + next(lines, "")
+            get_requirement(line)
+            return
+
+
+def deduce_helpful_msg(req: str) -> str:
+    """Returns helpful msg in case requirements file does not exist,
+    or cannot be parsed.
+
+    :params req: Requirements file path
+    """
+    if not os.path.exists(req):
+        return f" File '{req}' does not exist."
+    msg = " The path does exist. "
+    # Try to parse and check if it is a requirements file.
+    try:
+        check_first_requirement_in_file(req)
+    except InvalidRequirement:
+        logger.debug("Cannot parse '%s' as requirements file", req)
+    else:
+        msg += (
+            f"The argument you provided "
+            f"({req}) appears to be a"
+            f" requirements file. If that is the"
+            f" case, use the '-r' flag to install"
+            f" the packages specified within it."
+        )
+    return msg
+
+
+@dataclass(frozen=True)
+class RequirementParts:
+    requirement: Requirement | None
+    link: Link | None
+    markers: Marker | None
+    extras: set[str]
+
+
+def parse_req_from_editable(editable_req: str) -> RequirementParts:
+    name, url, extras_override = parse_editable(editable_req)
+
+    if name is not None:
+        try:
+            req: Requirement | None = get_requirement(name)
+        except InvalidRequirement as exc:
+            raise InstallationError(f"Invalid requirement: {name!r}: {exc}")
+    else:
+        req = None
+
+    link = Link(url)
+
+    return RequirementParts(req, link, None, extras_override)
+
+
+# ---- The actual constructors follow ----
+
+
+def install_req_from_editable(
+    editable_req: str,
+    comes_from: InstallRequirement | str | None = None,
+    *,
+    isolated: bool = False,
+    hash_options: dict[str, list[str]] | None = None,
+    constraint: bool = False,
+    user_supplied: bool = False,
+    permit_editable_wheels: bool = False,
+    config_settings: dict[str, str | list[str]] | None = None,
+) -> InstallRequirement:
+    if constraint:
+        raise InstallationError("Editable requirements are not allowed as constraints")
+
+    parts = parse_req_from_editable(editable_req)
+    return InstallRequirement(
+        parts.requirement,
+        comes_from=comes_from,
+        user_supplied=user_supplied,
+        editable=True,
+        permit_editable_wheels=permit_editable_wheels,
+        link=parts.link,
+        constraint=constraint,
+        isolated=isolated,
+        hash_options=hash_options,
+        config_settings=config_settings,
+        extras=parts.extras,
+    )
+
+
+def _looks_like_path(name: str) -> bool:
+    """Checks whether the string "looks like" a path on the filesystem.
+
+    This does not check whether the target actually exists, only judge from the
+    appearance.
+
+    Returns true if any of the following conditions is true:
+    * a path separator is found (either os.path.sep or os.path.altsep);
+    * a dot is found (which represents the current directory).
+    """
+    if os.path.sep in name:
+        return True
+    if os.path.altsep is not None and os.path.altsep in name:
+        return True
+    if name.startswith("."):
+        return True
+    return False
+
+
+def _get_url_from_path(path: str, name: str) -> str | None:
+    """
+    First, it checks whether a provided path is an installable directory. If it
+    is, returns the path.
+
+    If false, check if the path is an archive file (such as a .whl).
+    The function checks if the path is a file. If false, if the path has
+    an @, it will treat it as a PEP 440 URL requirement and return the path.
+    """
+    if _looks_like_path(name) and os.path.isdir(path):
+        if is_installable_dir(path):
+            return path_to_url(path)
+        # TODO: The is_installable_dir test here might not be necessary
+        #       now that it is done in load_pyproject_toml too.
+        raise InstallationError(
+            f"Directory {name!r} is not installable. Neither 'setup.py' "
+            "nor 'pyproject.toml' found."
+        )
+    if not is_archive_file(path):
+        return None
+    if os.path.isfile(path):
+        return path_to_url(path)
+    urlreq_parts = name.split("@", 1)
+    if len(urlreq_parts) >= 2 and not _looks_like_path(urlreq_parts[0]):
+        # If the path contains '@' and the part before it does not look
+        # like a path, try to treat it as a PEP 440 URL req instead.
+        return None
+    logger.warning(
+        "Requirement %r looks like a filename, but the file does not exist",
+        name,
+    )
+    return path_to_url(path)
+
+
+def parse_req_from_line(name: str, line_source: str | None) -> RequirementParts:
+    if is_url(name):
+        marker_sep = "; "
+    else:
+        marker_sep = ";"
+    if marker_sep in name:
+        name, markers_as_string = name.split(marker_sep, 1)
+        markers_as_string = markers_as_string.strip()
+        if not markers_as_string:
+            markers = None
+        else:
+            markers = Marker(markers_as_string)
+    else:
+        markers = None
+    name = name.strip()
+    req_as_string = None
+    path = os.path.normpath(os.path.abspath(name))
+    link = None
+    extras_as_string = None
+
+    if is_url(name):
+        link = Link(name)
+    else:
+        p, extras_as_string = _strip_extras(path)
+        url = _get_url_from_path(p, name)
+        if url is not None:
+            link = Link(url)
+
+    # it's a local file, dir, or url
+    if link:
+        # Handle relative file URLs
+        if link.scheme == "file" and re.search(r"\.\./", link.url):
+            link = Link(path_to_url(os.path.normpath(os.path.abspath(link.path))))
+        # wheel file
+        if link.is_wheel:
+            wheel = Wheel(link.filename)  # can raise InvalidWheelFilename
+            req_as_string = f"{wheel.name}=={wheel.version}"
+        else:
+            # set the req to the egg fragment.  when it's not there, this
+            # will become an 'unnamed' requirement
+            req_as_string = link.egg_fragment
+
+    # a requirement specifier
+    else:
+        req_as_string = name
+
+    extras = convert_extras(extras_as_string)
+
+    def with_source(text: str) -> str:
+        if not line_source:
+            return text
+        return f"{text} (from {line_source})"
+
+    def _parse_req_string(req_as_string: str) -> Requirement:
+        try:
+            return get_requirement(req_as_string)
+        except InvalidRequirement as exc:
+            if os.path.sep in req_as_string:
+                add_msg = "It looks like a path."
+                add_msg += deduce_helpful_msg(req_as_string)
+            elif "=" in req_as_string and not any(
+                op in req_as_string for op in operators
+            ):
+                add_msg = "= is not a valid operator. Did you mean == ?"
+            else:
+                add_msg = ""
+            msg = with_source(f"Invalid requirement: {req_as_string!r}: {exc}")
+            if add_msg:
+                msg += f"\nHint: {add_msg}"
+            raise InstallationError(msg)
+
+    if req_as_string is not None:
+        req: Requirement | None = _parse_req_string(req_as_string)
+    else:
+        req = None
+
+    return RequirementParts(req, link, markers, extras)
+
+
+def install_req_from_line(
+    name: str,
+    comes_from: str | InstallRequirement | None = None,
+    *,
+    isolated: bool = False,
+    hash_options: dict[str, list[str]] | None = None,
+    constraint: bool = False,
+    line_source: str | None = None,
+    user_supplied: bool = False,
+    config_settings: dict[str, str | list[str]] | None = None,
+) -> InstallRequirement:
+    """Creates an InstallRequirement from a name, which might be a
+    requirement, directory containing 'setup.py', filename, or URL.
+
+    :param line_source: An optional string describing where the line is from,
+        for logging purposes in case of an error.
+    """
+    parts = parse_req_from_line(name, line_source)
+
+    return InstallRequirement(
+        parts.requirement,
+        comes_from,
+        link=parts.link,
+        markers=parts.markers,
+        isolated=isolated,
+        hash_options=hash_options,
+        config_settings=config_settings,
+        constraint=constraint,
+        extras=parts.extras,
+        user_supplied=user_supplied,
+    )
+
+
+def install_req_from_req_string(
+    req_string: str,
+    comes_from: InstallRequirement | None = None,
+    isolated: bool = False,
+    user_supplied: bool = False,
+) -> InstallRequirement:
+    try:
+        req = get_requirement(req_string)
+    except InvalidRequirement as exc:
+        raise InstallationError(f"Invalid requirement: {req_string!r}: {exc}")
+
+    domains_not_allowed = [
+        PyPI.file_storage_domain,
+        TestPyPI.file_storage_domain,
+    ]
+    if (
+        req.url
+        and comes_from
+        and comes_from.link
+        and comes_from.link.netloc in domains_not_allowed
+    ):
+        # Explicitly disallow pypi packages that depend on external urls
+        raise InstallationError(
+            "Packages installed from PyPI cannot depend on packages "
+            "which are not also hosted on PyPI.\n"
+            f"{comes_from.name} depends on {req} "
+        )
+
+    return InstallRequirement(
+        req,
+        comes_from,
+        isolated=isolated,
+        user_supplied=user_supplied,
+    )
+
+
+def install_req_from_parsed_requirement(
+    parsed_req: ParsedRequirement,
+    isolated: bool = False,
+    user_supplied: bool = False,
+    config_settings: dict[str, str | list[str]] | None = None,
+) -> InstallRequirement:
+    if parsed_req.is_editable:
+        req = install_req_from_editable(
+            parsed_req.requirement,
+            comes_from=parsed_req.comes_from,
+            constraint=parsed_req.constraint,
+            isolated=isolated,
+            user_supplied=user_supplied,
+            config_settings=config_settings,
+        )
+
+    else:
+        req = install_req_from_line(
+            parsed_req.requirement,
+            comes_from=parsed_req.comes_from,
+            isolated=isolated,
+            hash_options=(
+                parsed_req.options.get("hashes", {}) if parsed_req.options else {}
+            ),
+            constraint=parsed_req.constraint,
+            line_source=parsed_req.line_source,
+            user_supplied=user_supplied,
+            config_settings=config_settings,
+        )
+    return req
+
+
+def install_req_from_link_and_ireq(
+    link: Link, ireq: InstallRequirement
+) -> InstallRequirement:
+    return InstallRequirement(
+        req=ireq.req,
+        comes_from=ireq.comes_from,
+        editable=ireq.editable,
+        link=link,
+        markers=ireq.markers,
+        isolated=ireq.isolated,
+        hash_options=ireq.hash_options,
+        config_settings=ireq.config_settings,
+        user_supplied=ireq.user_supplied,
+    )
+
+
+def install_req_drop_extras(ireq: InstallRequirement) -> InstallRequirement:
+    """
+    Creates a new InstallationRequirement using the given template but without
+    any extras. Sets the original requirement as the new one's parent
+    (comes_from).
+    """
+    return InstallRequirement(
+        req=(
+            _set_requirement_extras(ireq.req, set()) if ireq.req is not None else None
+        ),
+        comes_from=ireq,
+        editable=ireq.editable,
+        link=ireq.link,
+        markers=ireq.markers,
+        isolated=ireq.isolated,
+        hash_options=ireq.hash_options,
+        constraint=ireq.constraint,
+        extras=[],
+        config_settings=ireq.config_settings,
+        user_supplied=ireq.user_supplied,
+        permit_editable_wheels=ireq.permit_editable_wheels,
+    )
+
+
+def install_req_extend_extras(
+    ireq: InstallRequirement,
+    extras: Collection[str],
+) -> InstallRequirement:
+    """
+    Returns a copy of an installation requirement with some additional extras.
+    Makes a shallow copy of the ireq object.
+    """
+    result = copy.copy(ireq)
+    result.extras = {*ireq.extras, *extras}
+    result.req = (
+        _set_requirement_extras(ireq.req, result.extras)
+        if ireq.req is not None
+        else None
+    )
+    return result

venv/Lib/site-packages/pip/_internal/req/pep723.py

@@ -0,0 +1,41 @@
+import re
+from typing import Any
+
+from pip._internal.utils.compat import tomllib
+
+REGEX = r"(?m)^# /// (?P<type>[a-zA-Z0-9-]+)$\s(?P<content>(^#(| .*)$\s)+)^# ///$"
+
+
+class PEP723Exception(ValueError):
+    """Raised to indicate a problem when parsing PEP 723 metadata from a script"""
+
+    def __init__(self, msg: str) -> None:
+        self.msg = msg
+
+
+def pep723_metadata(scriptfile: str) -> dict[str, Any]:
+    with open(scriptfile) as f:
+        script = f.read()
+
+    name = "script"
+    matches = list(
+        filter(lambda m: m.group("type") == name, re.finditer(REGEX, script))
+    )
+
+    if len(matches) > 1:
+        raise PEP723Exception(f"Multiple {name!r} blocks found in {scriptfile!r}")
+    elif len(matches) == 1:
+        content = "".join(
+            line[2:] if line.startswith("# ") else line[1:]
+            for line in matches[0].group("content").splitlines(keepends=True)
+        )
+        try:
+            metadata = tomllib.loads(content)
+        except Exception as exc:
+            raise PEP723Exception(f"Failed to parse TOML in {scriptfile!r}") from exc
+    else:
+        raise PEP723Exception(
+            f"File does not contain {name!r} metadata: {scriptfile!r}"
+        )
+
+    return metadata

venv/Lib/site-packages/pip/_internal/req/req_dependency_group.py

@@ -0,0 +1,75 @@
+from collections.abc import Iterable, Iterator
+from typing import Any
+
+from pip._vendor.dependency_groups import DependencyGroupResolver
+
+from pip._internal.exceptions import InstallationError
+from pip._internal.utils.compat import tomllib
+
+
+def parse_dependency_groups(groups: list[tuple[str, str]]) -> list[str]:
+    """
+    Parse dependency groups data as provided via the CLI, in a `[path:]group` syntax.
+
+    Raises InstallationErrors if anything goes wrong.
+    """
+    resolvers = _build_resolvers(path for (path, _) in groups)
+    return list(_resolve_all_groups(resolvers, groups))
+
+
+def _resolve_all_groups(
+    resolvers: dict[str, DependencyGroupResolver], groups: list[tuple[str, str]]
+) -> Iterator[str]:
+    """
+    Run all resolution, converting any error from `DependencyGroupResolver` into
+    an InstallationError.
+    """
+    for path, groupname in groups:
+        resolver = resolvers[path]
+        try:
+            yield from (str(req) for req in resolver.resolve(groupname))
+        except (ValueError, TypeError, LookupError) as e:
+            raise InstallationError(
+                f"[dependency-groups] resolution failed for '{groupname}' "
+                f"from '{path}': {e}"
+            ) from e
+
+
+def _build_resolvers(paths: Iterable[str]) -> dict[str, Any]:
+    resolvers = {}
+    for path in paths:
+        if path in resolvers:
+            continue
+
+        pyproject = _load_pyproject(path)
+        if "dependency-groups" not in pyproject:
+            raise InstallationError(
+                f"[dependency-groups] table was missing from '{path}'. "
+                "Cannot resolve '--group' option."
+            )
+        raw_dependency_groups = pyproject["dependency-groups"]
+        if not isinstance(raw_dependency_groups, dict):
+            raise InstallationError(
+                f"[dependency-groups] table was malformed in {path}. "
+                "Cannot resolve '--group' option."
+            )
+
+        resolvers[path] = DependencyGroupResolver(raw_dependency_groups)
+    return resolvers
+
+
+def _load_pyproject(path: str) -> dict[str, Any]:
+    """
+    This helper loads a pyproject.toml as TOML.
+
+    It raises an InstallationError if the operation fails.
+    """
+    try:
+        with open(path, "rb") as fp:
+            return tomllib.load(fp)
+    except FileNotFoundError:
+        raise InstallationError(f"{path} not found. Cannot resolve '--group' option.")
+    except tomllib.TOMLDecodeError as e:
+        raise InstallationError(f"Error parsing {path}: {e}") from e
+    except OSError as e:
+        raise InstallationError(f"Error reading {path}: {e}") from e

venv/Lib/site-packages/pip/_internal/req/req_file.py

@@ -0,0 +1,631 @@
+"""
+Requirements file parsing
+"""
+
+from __future__ import annotations
+
+import codecs
+import locale
+import logging
+import optparse
+import os
+import re
+import shlex
+import sys
+import urllib.parse
+from collections.abc import Generator, Iterable
+from dataclasses import dataclass
+from optparse import Values
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    NoReturn,
+)
+
+from pip._internal.cli import cmdoptions
+from pip._internal.exceptions import InstallationError, RequirementsFileParseError
+from pip._internal.models.release_control import ReleaseControl
+from pip._internal.models.search_scope import SearchScope
+
+if TYPE_CHECKING:
+    from pip._internal.index.package_finder import PackageFinder
+    from pip._internal.network.session import PipSession
+
+__all__ = ["parse_requirements"]
+
+ReqFileLines = Iterable[tuple[int, str]]
+
+LineParser = Callable[[str], tuple[str, Values]]
+
+SCHEME_RE = re.compile(r"^(http|https|file):", re.I)
+COMMENT_RE = re.compile(r"(^|\s+)#.*$")
+
+# Matches environment variable-style values in '${MY_VARIABLE_1}' with the
+# variable name consisting of only uppercase letters, digits or the '_'
+# (underscore). This follows the POSIX standard defined in IEEE Std 1003.1,
+# 2013 Edition.
+ENV_VAR_RE = re.compile(r"(?P<var>\$\{(?P<name>[A-Z0-9_]+)\})")
+
+SUPPORTED_OPTIONS: list[Callable[..., optparse.Option]] = [
+    cmdoptions.index_url,
+    cmdoptions.extra_index_url,
+    cmdoptions.no_index,
+    cmdoptions.constraints,
+    cmdoptions.requirements,
+    cmdoptions.editable,
+    cmdoptions.find_links,
+    cmdoptions.no_binary,
+    cmdoptions.only_binary,
+    cmdoptions.prefer_binary,
+    cmdoptions.require_hashes,
+    cmdoptions.pre,
+    cmdoptions.all_releases,
+    cmdoptions.only_final,
+    cmdoptions.trusted_host,
+    cmdoptions.use_new_feature,
+]
+
+# options to be passed to requirements
+SUPPORTED_OPTIONS_REQ: list[Callable[..., optparse.Option]] = [
+    cmdoptions.hash,
+    cmdoptions.config_settings,
+]
+
+SUPPORTED_OPTIONS_EDITABLE_REQ: list[Callable[..., optparse.Option]] = [
+    cmdoptions.config_settings,
+]
+
+
+# the 'dest' string values
+SUPPORTED_OPTIONS_REQ_DEST = [str(o().dest) for o in SUPPORTED_OPTIONS_REQ]
+SUPPORTED_OPTIONS_EDITABLE_REQ_DEST = [
+    str(o().dest) for o in SUPPORTED_OPTIONS_EDITABLE_REQ
+]
+
+# order of BOMS is important: codecs.BOM_UTF16_LE is a prefix of codecs.BOM_UTF32_LE
+# so data.startswith(BOM_UTF16_LE) would be true for UTF32_LE data
+BOMS: list[tuple[bytes, str]] = [
+    (codecs.BOM_UTF8, "utf-8"),
+    (codecs.BOM_UTF32, "utf-32"),
+    (codecs.BOM_UTF32_BE, "utf-32-be"),
+    (codecs.BOM_UTF32_LE, "utf-32-le"),
+    (codecs.BOM_UTF16, "utf-16"),
+    (codecs.BOM_UTF16_BE, "utf-16-be"),
+    (codecs.BOM_UTF16_LE, "utf-16-le"),
+]
+
+PEP263_ENCODING_RE = re.compile(rb"coding[:=]\s*([-\w.]+)")
+DEFAULT_ENCODING = "utf-8"
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ParsedRequirement:
+    # TODO: replace this with slots=True when dropping Python 3.9 support.
+    __slots__ = (
+        "requirement",
+        "is_editable",
+        "comes_from",
+        "constraint",
+        "options",
+        "line_source",
+    )
+
+    requirement: str
+    is_editable: bool
+    comes_from: str
+    constraint: bool
+    options: dict[str, Any] | None
+    line_source: str | None
+
+
+@dataclass(frozen=True)
+class ParsedLine:
+    __slots__ = ("filename", "lineno", "args", "opts", "constraint")
+
+    filename: str
+    lineno: int
+    args: str
+    opts: Values
+    constraint: bool
+
+    @property
+    def is_editable(self) -> bool:
+        return bool(self.opts.editables)
+
+    @property
+    def requirement(self) -> str | None:
+        if self.args:
+            return self.args
+        elif self.is_editable:
+            # We don't support multiple -e on one line
+            return self.opts.editables[0]
+        return None
+
+
+def parse_requirements(
+    filename: str,
+    session: PipSession,
+    finder: PackageFinder | None = None,
+    options: optparse.Values | None = None,
+    constraint: bool = False,
+) -> Generator[ParsedRequirement, None, None]:
+    """Parse a requirements file and yield ParsedRequirement instances.
+
+    :param filename:    Path or url of requirements file.
+    :param session:     PipSession instance.
+    :param finder:      Instance of pip.index.PackageFinder.
+    :param options:     cli options.
+    :param constraint:  If true, parsing a constraint file rather than
+        requirements file.
+    """
+    line_parser = get_line_parser(finder)
+    parser = RequirementsFileParser(session, line_parser)
+
+    for parsed_line in parser.parse(filename, constraint):
+        parsed_req = handle_line(
+            parsed_line, options=options, finder=finder, session=session
+        )
+        if parsed_req is not None:
+            yield parsed_req
+
+
+def preprocess(content: str) -> ReqFileLines:
+    """Split, filter, and join lines, and return a line iterator
+
+    :param content: the content of the requirements file
+    """
+    lines_enum: ReqFileLines = enumerate(content.splitlines(), start=1)
+    lines_enum = join_lines(lines_enum)
+    lines_enum = ignore_comments(lines_enum)
+    lines_enum = expand_env_variables(lines_enum)
+    return lines_enum
+
+
+def handle_requirement_line(
+    line: ParsedLine,
+    options: optparse.Values | None = None,
+) -> ParsedRequirement:
+    # preserve for the nested code path
+    line_comes_from = "{} {} (line {})".format(
+        "-c" if line.constraint else "-r",
+        line.filename,
+        line.lineno,
+    )
+
+    assert line.requirement is not None
+
+    # get the options that apply to requirements
+    if line.is_editable:
+        supported_dest = SUPPORTED_OPTIONS_EDITABLE_REQ_DEST
+    else:
+        supported_dest = SUPPORTED_OPTIONS_REQ_DEST
+    req_options = {}
+    for dest in supported_dest:
+        if dest in line.opts.__dict__ and line.opts.__dict__[dest]:
+            req_options[dest] = line.opts.__dict__[dest]
+
+    line_source = f"line {line.lineno} of {line.filename}"
+    return ParsedRequirement(
+        requirement=line.requirement,
+        is_editable=line.is_editable,
+        comes_from=line_comes_from,
+        constraint=line.constraint,
+        options=req_options,
+        line_source=line_source,
+    )
+
+
+def handle_option_line(
+    opts: Values,
+    filename: str,
+    lineno: int,
+    finder: PackageFinder | None = None,
+    options: optparse.Values | None = None,
+    session: PipSession | None = None,
+) -> None:
+    if opts.hashes:
+        logger.warning(
+            "%s line %s has --hash but no requirement, and will be ignored.",
+            filename,
+            lineno,
+        )
+
+    if options:
+        # percolate options upward
+        if opts.require_hashes:
+            options.require_hashes = opts.require_hashes
+        if opts.features_enabled:
+            options.features_enabled.extend(
+                f for f in opts.features_enabled if f not in options.features_enabled
+            )
+
+    # set finder options
+    if finder:
+        find_links = finder.find_links
+        index_urls = finder.index_urls
+        no_index = finder.search_scope.no_index
+        if opts.no_index is True:
+            no_index = True
+            index_urls = []
+        if opts.index_url and not no_index:
+            index_urls = [opts.index_url]
+        if opts.extra_index_urls and not no_index:
+            index_urls.extend(opts.extra_index_urls)
+        if opts.find_links:
+            # FIXME: it would be nice to keep track of the source
+            # of the find_links: support a find-links local path
+            # relative to a requirements file.
+            value = opts.find_links[0]
+            req_dir = os.path.dirname(os.path.abspath(filename))
+            relative_to_reqs_file = os.path.join(req_dir, value)
+            if os.path.exists(relative_to_reqs_file):
+                value = relative_to_reqs_file
+            find_links.append(value)
+
+        if session:
+            # We need to update the auth urls in session
+            session.update_index_urls(index_urls)
+
+        search_scope = SearchScope(
+            find_links=find_links,
+            index_urls=index_urls,
+            no_index=no_index,
+        )
+        finder.search_scope = search_scope
+
+        # Transform --pre into --all-releases :all:
+        if opts.pre:
+            if not opts.release_control:
+                opts.release_control = ReleaseControl()
+            opts.release_control.all_releases.add(":all:")
+
+        if opts.release_control:
+            if not finder.release_control:
+                # First time seeing release_control, set it on finder
+                finder.set_release_control(opts.release_control)
+
+        if opts.prefer_binary:
+            finder.set_prefer_binary()
+
+        if session:
+            for host in opts.trusted_hosts or []:
+                source = f"line {lineno} of {filename}"
+                session.add_trusted_host(host, source=source)
+
+
+def handle_line(
+    line: ParsedLine,
+    options: optparse.Values | None = None,
+    finder: PackageFinder | None = None,
+    session: PipSession | None = None,
+) -> ParsedRequirement | None:
+    """Handle a single parsed requirements line; This can result in
+    creating/yielding requirements, or updating the finder.
+
+    :param line:        The parsed line to be processed.
+    :param options:     CLI options.
+    :param finder:      The finder - updated by non-requirement lines.
+    :param session:     The session - updated by non-requirement lines.
+
+    Returns a ParsedRequirement object if the line is a requirement line,
+    otherwise returns None.
+
+    For lines that contain requirements, the only options that have an effect
+    are from SUPPORTED_OPTIONS_REQ, and they are scoped to the
+    requirement. Other options from SUPPORTED_OPTIONS may be present, but are
+    ignored.
+
+    For lines that do not contain requirements, the only options that have an
+    effect are from SUPPORTED_OPTIONS. Options from SUPPORTED_OPTIONS_REQ may
+    be present, but are ignored. These lines may contain multiple options
+    (although our docs imply only one is supported), and all our parsed and
+    affect the finder.
+    """
+
+    if line.requirement is not None:
+        parsed_req = handle_requirement_line(line, options)
+        return parsed_req
+    else:
+        handle_option_line(
+            line.opts,
+            line.filename,
+            line.lineno,
+            finder,
+            options,
+            session,
+        )
+        return None
+
+
+class RequirementsFileParser:
+    def __init__(
+        self,
+        session: PipSession,
+        line_parser: LineParser,
+    ) -> None:
+        self._session = session
+        self._line_parser = line_parser
+
+    def parse(
+        self, filename: str, constraint: bool
+    ) -> Generator[ParsedLine, None, None]:
+        """Parse a given file, yielding parsed lines."""
+        yield from self._parse_and_recurse(
+            filename, constraint, [{os.path.abspath(filename): None}]
+        )
+
+    def _parse_and_recurse(
+        self,
+        filename: str,
+        constraint: bool,
+        parsed_files_stack: list[dict[str, str | None]],
+    ) -> Generator[ParsedLine, None, None]:
+        for line in self._parse_file(filename, constraint):
+            if line.requirement is None and (
+                line.opts.requirements or line.opts.constraints
+            ):
+                # parse a nested requirements file
+                if line.opts.requirements:
+                    req_path = line.opts.requirements[0]
+                    nested_constraint = False
+                else:
+                    req_path = line.opts.constraints[0]
+                    nested_constraint = True
+
+                # original file is over http
+                if SCHEME_RE.search(filename):
+                    # do a url join so relative paths work
+                    req_path = urllib.parse.urljoin(filename, req_path)
+                # original file and nested file are paths
+                elif not SCHEME_RE.search(req_path):
+                    # do a join so relative paths work
+                    # and then abspath so that we can identify recursive references
+                    req_path = os.path.abspath(
+                        os.path.join(
+                            os.path.dirname(filename),
+                            req_path,
+                        )
+                    )
+                parsed_files = parsed_files_stack[0]
+                if req_path in parsed_files:
+                    initial_file = parsed_files[req_path]
+                    tail = (
+                        f" and again in {initial_file}"
+                        if initial_file is not None
+                        else ""
+                    )
+                    raise RequirementsFileParseError(
+                        f"{req_path} recursively references itself in {filename}{tail}"
+                    )
+                # Keeping a track where was each file first included in
+                new_parsed_files = parsed_files.copy()
+                new_parsed_files[req_path] = filename
+                yield from self._parse_and_recurse(
+                    req_path, nested_constraint, [new_parsed_files, *parsed_files_stack]
+                )
+            else:
+                yield line
+
+    def _parse_file(
+        self, filename: str, constraint: bool
+    ) -> Generator[ParsedLine, None, None]:
+        _, content = get_file_content(filename, self._session)
+
+        lines_enum = preprocess(content)
+
+        for line_number, line in lines_enum:
+            try:
+                args_str, opts = self._line_parser(line)
+            except OptionParsingError as e:
+                # add offending line
+                msg = f"Invalid requirement: {line}\n{e.msg}"
+                raise RequirementsFileParseError(msg)
+
+            yield ParsedLine(
+                filename,
+                line_number,
+                args_str,
+                opts,
+                constraint,
+            )
+
+
+def get_line_parser(finder: PackageFinder | None) -> LineParser:
+    def parse_line(line: str) -> tuple[str, Values]:
+        # Build new parser for each line since it accumulates appendable
+        # options.
+        parser = build_parser()
+        defaults = parser.get_default_values()
+        defaults.index_url = None
+        if finder:
+            defaults.format_control = finder.format_control
+            defaults.release_control = finder.release_control
+
+        args_str, options_str = break_args_options(line)
+
+        try:
+            options = shlex.split(options_str)
+        except ValueError as e:
+            raise OptionParsingError(f"Could not split options: {options_str}") from e
+
+        opts, _ = parser.parse_args(options, defaults)
+
+        return args_str, opts
+
+    return parse_line
+
+
+def break_args_options(line: str) -> tuple[str, str]:
+    """Break up the line into an args and options string.  We only want to shlex
+    (and then optparse) the options, not the args.  args can contain markers
+    which are corrupted by shlex.
+    """
+    tokens = line.split(" ")
+    args = []
+    options = tokens[:]
+    for token in tokens:
+        if token.startswith(("-", "--")):
+            break
+        else:
+            args.append(token)
+            options.pop(0)
+    return " ".join(args), " ".join(options)
+
+
+class OptionParsingError(Exception):
+    def __init__(self, msg: str) -> None:
+        self.msg = msg
+
+
+def build_parser() -> optparse.OptionParser:
+    """
+    Return a parser for parsing requirement lines
+    """
+    parser = optparse.OptionParser(add_help_option=False)
+
+    option_factories = SUPPORTED_OPTIONS + SUPPORTED_OPTIONS_REQ
+    for option_factory in option_factories:
+        option = option_factory()
+        parser.add_option(option)
+
+    # By default optparse sys.exits on parsing errors. We want to wrap
+    # that in our own exception.
+    def parser_exit(self: Any, msg: str) -> NoReturn:
+        raise OptionParsingError(msg)
+
+    # NOTE: mypy disallows assigning to a method
+    #       https://github.com/python/mypy/issues/2427
+    parser.exit = parser_exit  # type: ignore
+
+    return parser
+
+
+def join_lines(lines_enum: ReqFileLines) -> ReqFileLines:
+    """Joins a line ending in '\' with the previous line (except when following
+    comments).  The joined line takes on the index of the first line.
+    """
+    primary_line_number = None
+    new_line: list[str] = []
+    for line_number, line in lines_enum:
+        if not line.endswith("\\") or COMMENT_RE.match(line):
+            if COMMENT_RE.match(line):
+                # this ensures comments are always matched later
+                line = " " + line
+            if new_line:
+                new_line.append(line)
+                assert primary_line_number is not None
+                yield primary_line_number, "".join(new_line)
+                new_line = []
+            else:
+                yield line_number, line
+        else:
+            if not new_line:
+                primary_line_number = line_number
+            new_line.append(line.strip("\\"))
+
+    # last line contains \
+    if new_line:
+        assert primary_line_number is not None
+        yield primary_line_number, "".join(new_line)
+
+    # TODO: handle space after '\'.
+
+
+def ignore_comments(lines_enum: ReqFileLines) -> ReqFileLines:
+    """
+    Strips comments and filter empty lines.
+    """
+    for line_number, line in lines_enum:
+        line = COMMENT_RE.sub("", line)
+        line = line.strip()
+        if line:
+            yield line_number, line
+
+
+def expand_env_variables(lines_enum: ReqFileLines) -> ReqFileLines:
+    """Replace all environment variables that can be retrieved via `os.getenv`.
+
+    The only allowed format for environment variables defined in the
+    requirement file is `${MY_VARIABLE_1}` to ensure two things:
+
+    1. Strings that contain a `$` aren't accidentally (partially) expanded.
+    2. Ensure consistency across platforms for requirement files.
+
+    These points are the result of a discussion on the `github pull
+    request #3514 <https://github.com/pypa/pip/pull/3514>`_.
+
+    Valid characters in variable names follow the `POSIX standard
+    <http://pubs.opengroup.org/onlinepubs/9699919799/>`_ and are limited
+    to uppercase letter, digits and the `_` (underscore).
+    """
+    for line_number, line in lines_enum:
+        for env_var, var_name in ENV_VAR_RE.findall(line):
+            value = os.getenv(var_name)
+            if not value:
+                continue
+
+            line = line.replace(env_var, value)
+
+        yield line_number, line
+
+
+def get_file_content(url: str, session: PipSession) -> tuple[str, str]:
+    """Gets the content of a file; it may be a filename, file: URL, or
+    http: URL.  Returns (location, content).  Content is unicode.
+    Respects # -*- coding: declarations on the retrieved files.
+
+    :param url:         File path or url.
+    :param session:     PipSession instance.
+    """
+    scheme = urllib.parse.urlsplit(url).scheme
+    # Pip has special support for file:// URLs (LocalFSAdapter).
+    if scheme in ["http", "https", "file"]:
+        # Delay importing heavy network modules until absolutely necessary.
+        from pip._internal.network.utils import raise_for_status
+
+        resp = session.get(url)
+        raise_for_status(resp)
+        return resp.url, resp.text
+
+    # Assume this is a bare path.
+    try:
+        with open(url, "rb") as f:
+            raw_content = f.read()
+    except OSError as exc:
+        raise InstallationError(f"Could not open requirements file: {exc}")
+
+    content = _decode_req_file(raw_content, url)
+
+    return url, content
+
+
+def _decode_req_file(data: bytes, url: str) -> str:
+    for bom, encoding in BOMS:
+        if data.startswith(bom):
+            return data[len(bom) :].decode(encoding)
+
+    for line in data.split(b"\n")[:2]:
+        if line[0:1] == b"#":
+            result = PEP263_ENCODING_RE.search(line)
+            if result is not None:
+                encoding = result.groups()[0].decode("ascii")
+                return data.decode(encoding)
+
+    try:
+        return data.decode(DEFAULT_ENCODING)
+    except UnicodeDecodeError:
+        locale_encoding = locale.getpreferredencoding(False) or sys.getdefaultencoding()
+        logging.warning(
+            "unable to decode data from %s with default encoding %s, "
+            "falling back to encoding from locale: %s. "
+            "If this is intentional you should specify the encoding with a "
+            "PEP-263 style comment, e.g. '# -*- coding: %s -*-'",
+            url,
+            DEFAULT_ENCODING,
+            locale_encoding,
+            locale_encoding,
+        )
+        return data.decode(locale_encoding)