import collections import logging from typing import Generator, List, Optional, Sequence, Tuple 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__) class InstallationResult: def __init__(self, name: str) -> None: self.name = name def __repr__(self) -> str: return f"InstallationResult(name={self.name!r})" 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], install_options: List[str], global_options: Sequence[str], root: Optional[str], home: Optional[str], prefix: Optional[str], warn_script_location: bool, use_user_site: bool, pycompile: bool, ) -> 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 = [] with indent_log(): for req_name, requirement in to_install.items(): 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( install_options, global_options, 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