| import logging | |
| from collections import OrderedDict | |
| from typing import Dict, Iterable, List, Optional, Tuple | |
| from pip._vendor.packaging.utils import canonicalize_name | |
| from pip._internal.exceptions import InstallationError | |
| from pip._internal.models.wheel import Wheel | |
| from pip._internal.req.req_install import InstallRequirement | |
| from pip._internal.utils import compatibility_tags | |
| logger = logging.getLogger(__name__) | |
| class RequirementSet: | |
| def __init__(self, check_supported_wheels: bool = True) -> None: | |
| """Create a RequirementSet.""" | |
| self.requirements: Dict[str, InstallRequirement] = OrderedDict() | |
| self.check_supported_wheels = check_supported_wheels | |
| self.unnamed_requirements: List[InstallRequirement] = [] | |
| def __str__(self) -> str: | |
| requirements = sorted( | |
| (req for req in self.requirements.values() if not req.comes_from), | |
| key=lambda req: canonicalize_name(req.name or ""), | |
| ) | |
| return " ".join(str(req.req) for req in requirements) | |
| def __repr__(self) -> str: | |
| requirements = sorted( | |
| self.requirements.values(), | |
| key=lambda req: canonicalize_name(req.name or ""), | |
| ) | |
| format_string = "<{classname} object; {count} requirement(s): {reqs}>" | |
| return format_string.format( | |
| classname=self.__class__.__name__, | |
| count=len(requirements), | |
| reqs=", ".join(str(req.req) for req in requirements), | |
| ) | |
| def add_unnamed_requirement(self, install_req: InstallRequirement) -> None: | |
| assert not install_req.name | |
| self.unnamed_requirements.append(install_req) | |
| def add_named_requirement(self, install_req: InstallRequirement) -> None: | |
| assert install_req.name | |
| project_name = canonicalize_name(install_req.name) | |
| self.requirements[project_name] = install_req | |
| def add_requirement( | |
| self, | |
| install_req: InstallRequirement, | |
| parent_req_name: Optional[str] = None, | |
| extras_requested: Optional[Iterable[str]] = None, | |
| ) -> Tuple[List[InstallRequirement], Optional[InstallRequirement]]: | |
| """Add install_req as a requirement to install. | |
| :param parent_req_name: The name of the requirement that needed this | |
| added. The name is used because when multiple unnamed requirements | |
| resolve to the same name, we could otherwise end up with dependency | |
| links that point outside the Requirements set. parent_req must | |
| already be added. Note that None implies that this is a user | |
| supplied requirement, vs an inferred one. | |
| :param extras_requested: an iterable of extras used to evaluate the | |
| environment markers. | |
| :return: Additional requirements to scan. That is either [] if | |
| the requirement is not applicable, or [install_req] if the | |
| requirement is applicable and has just been added. | |
| """ | |
| # If the markers do not match, ignore this requirement. | |
| if not install_req.match_markers(extras_requested): | |
| logger.info( | |
| "Ignoring %s: markers '%s' don't match your environment", | |
| install_req.name, | |
| install_req.markers, | |
| ) | |
| return [], None | |
| # If the wheel is not supported, raise an error. | |
| # Should check this after filtering out based on environment markers to | |
| # allow specifying different wheels based on the environment/OS, in a | |
| # single requirements file. | |
| if install_req.link and install_req.link.is_wheel: | |
| wheel = Wheel(install_req.link.filename) | |
| tags = compatibility_tags.get_supported() | |
| if self.check_supported_wheels and not wheel.supported(tags): | |
| raise InstallationError( | |
| "{} is not a supported wheel on this platform.".format( | |
| wheel.filename | |
| ) | |
| ) | |
| # This next bit is really a sanity check. | |
| assert ( | |
| not install_req.user_supplied or parent_req_name is None | |
| ), "a user supplied req shouldn't have a parent" | |
| # Unnamed requirements are scanned again and the requirement won't be | |
| # added as a dependency until after scanning. | |
| if not install_req.name: | |
| self.add_unnamed_requirement(install_req) | |
| return [install_req], None | |
| try: | |
| existing_req: Optional[InstallRequirement] = self.get_requirement( | |
| install_req.name | |
| ) | |
| except KeyError: | |
| existing_req = None | |
| has_conflicting_requirement = ( | |
| parent_req_name is None | |
| and existing_req | |
| and not existing_req.constraint | |
| and existing_req.extras == install_req.extras | |
| and existing_req.req | |
| and install_req.req | |
| and existing_req.req.specifier != install_req.req.specifier | |
| ) | |
| if has_conflicting_requirement: | |
| raise InstallationError( | |
| "Double requirement given: {} (already in {}, name={!r})".format( | |
| install_req, existing_req, install_req.name | |
| ) | |
| ) | |
| # When no existing requirement exists, add the requirement as a | |
| # dependency and it will be scanned again after. | |
| if not existing_req: | |
| self.add_named_requirement(install_req) | |
| # We'd want to rescan this requirement later | |
| return [install_req], install_req | |
| # Assume there's no need to scan, and that we've already | |
| # encountered this for scanning. | |
| if install_req.constraint or not existing_req.constraint: | |
| return [], existing_req | |
| does_not_satisfy_constraint = install_req.link and not ( | |
| existing_req.link and install_req.link.path == existing_req.link.path | |
| ) | |
| if does_not_satisfy_constraint: | |
| raise InstallationError( | |
| "Could not satisfy constraints for '{}': " | |
| "installation from path or url cannot be " | |
| "constrained to a version".format(install_req.name) | |
| ) | |
| # If we're now installing a constraint, mark the existing | |
| # object for real installation. | |
| existing_req.constraint = False | |
| # If we're now installing a user supplied requirement, | |
| # mark the existing object as such. | |
| if install_req.user_supplied: | |
| existing_req.user_supplied = True | |
| existing_req.extras = tuple( | |
| sorted(set(existing_req.extras) | set(install_req.extras)) | |
| ) | |
| logger.debug( | |
| "Setting %s extras to: %s", | |
| existing_req, | |
| existing_req.extras, | |
| ) | |
| # Return the existing requirement for addition to the parent and | |
| # scanning again. | |
| return [existing_req], existing_req | |
| def has_requirement(self, name: str) -> bool: | |
| project_name = canonicalize_name(name) | |
| return ( | |
| project_name in self.requirements | |
| and not self.requirements[project_name].constraint | |
| ) | |
| def get_requirement(self, name: str) -> InstallRequirement: | |
| project_name = canonicalize_name(name) | |
| if project_name in self.requirements: | |
| return self.requirements[project_name] | |
| raise KeyError(f"No project with the name {name!r}") | |
| @property | |
| def all_requirements(self) -> List[InstallRequirement]: | |
| return self.unnamed_requirements + list(self.requirements.values()) |