from __future__ import absolute_import import logging from collections import OrderedDict from pip._internal.exceptions import InstallationError from pip._internal.utils.logging import indent_log from pip._internal.utils.typing import MYPY_CHECK_RUNNING from pip._internal.wheel import Wheel if MYPY_CHECK_RUNNING: from typing import Dict, Iterable, List, Optional, Tuple from pip._internal.req.req_install import InstallRequirement logger = logging.getLogger(__name__) class RequirementSet(object): def __init__(self, require_hashes=False, check_supported_wheels=True): # type: (bool, bool) -> None """Create a RequirementSet. """ self.requirements = OrderedDict() # type: Dict[str, InstallRequirement] # noqa: E501 self.require_hashes = require_hashes self.check_supported_wheels = check_supported_wheels # Mapping of alias: real_name self.requirement_aliases = {} # type: Dict[str, str] self.unnamed_requirements = [] # type: List[InstallRequirement] self.successfully_downloaded = [] # type: List[InstallRequirement] self.reqs_to_cleanup = [] # type: List[InstallRequirement] def __str__(self): # type: () -> str reqs = [req for req in self.requirements.values() if not req.comes_from] reqs.sort(key=lambda req: req.name.lower()) return ' '.join([str(req.req) for req in reqs]) def __repr__(self): # type: () -> str reqs = [req for req in self.requirements.values()] reqs.sort(key=lambda req: req.name.lower()) reqs_str = ', '.join([str(req.req) for req in reqs]) return ('<%s object; %d requirement(s): %s>' % (self.__class__.__name__, len(reqs), reqs_str)) def add_requirement( self, install_req, # type: InstallRequirement parent_req_name=None, # type: Optional[str] extras_requested=None # type: Optional[Iterable[str]] ): # type: (...) -> Tuple[List[InstallRequirement], Optional[InstallRequirement]] # noqa: E501 """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. """ name = install_req.name # 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", 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) if self.check_supported_wheels and not wheel.supported(): raise InstallationError( "%s is not a supported wheel on this platform." % wheel.filename ) # This next bit is really a sanity check. assert install_req.is_direct == (parent_req_name is None), ( "a direct req shouldn't have a parent and also, " "a non direct req should have a parent" ) # Unnamed requirements are scanned again and the requirement won't be # added as a dependency until after scanning. if not name: # url or path requirement w/o an egg fragment self.unnamed_requirements.append(install_req) return [install_req], None try: existing_req = self.get_requirement(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.specifier != install_req.req.specifier ) if has_conflicting_requirement: raise InstallationError( "Double requirement given: %s (already in %s, name=%r)" % (install_req, existing_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.requirements[name] = install_req # FIXME: what about other normalizations? E.g., _ vs. -? if name.lower() != name: self.requirement_aliases[name.lower()] = name # We'd want to rescan this requirements 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: self.reqs_to_cleanup.append(install_req) raise InstallationError( "Could not satisfy constraints for '%s': " "installation from path or url cannot be " "constrained to a version" % name, ) # If we're now installing a constraint, mark the existing # object for real installation. existing_req.constraint = False 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, project_name): # type: (str) -> bool name = project_name.lower() if (name in self.requirements and not self.requirements[name].constraint or name in self.requirement_aliases and not self.requirements[self.requirement_aliases[name]].constraint): return True return False def get_requirement(self, project_name): # type: (str) -> InstallRequirement for name in project_name, project_name.lower(): if name in self.requirements: return self.requirements[name] if name in self.requirement_aliases: return self.requirements[self.requirement_aliases[name]] raise KeyError("No project with the name %r" % project_name) def cleanup_files(self): # type: () -> None """Clean up files, remove builds.""" logger.debug('Cleaning up...') with indent_log(): for req in self.reqs_to_cleanup: req.remove_temporary_source()