Yi Kong | 878f994 | 2023-12-13 12:55:04 +0900 | [diff] [blame^] | 1 | import collections |
| 2 | import logging |
| 3 | from typing import Iterator, List, Optional, Sequence, Tuple |
| 4 | |
| 5 | from pip._internal.utils.logging import indent_log |
| 6 | |
| 7 | from .req_file import parse_requirements |
| 8 | from .req_install import InstallRequirement |
| 9 | from .req_set import RequirementSet |
| 10 | |
| 11 | __all__ = [ |
| 12 | "RequirementSet", |
| 13 | "InstallRequirement", |
| 14 | "parse_requirements", |
| 15 | "install_given_reqs", |
| 16 | ] |
| 17 | |
| 18 | logger = logging.getLogger(__name__) |
| 19 | |
| 20 | |
| 21 | class InstallationResult: |
| 22 | def __init__(self, name: str) -> None: |
| 23 | self.name = name |
| 24 | |
| 25 | def __repr__(self) -> str: |
| 26 | return f"InstallationResult(name={self.name!r})" |
| 27 | |
| 28 | |
| 29 | def _validate_requirements( |
| 30 | requirements: List[InstallRequirement], |
| 31 | ) -> Iterator[Tuple[str, InstallRequirement]]: |
| 32 | for req in requirements: |
| 33 | assert req.name, f"invalid to-be-installed requirement: {req}" |
| 34 | yield req.name, req |
| 35 | |
| 36 | |
| 37 | def install_given_reqs( |
| 38 | requirements: List[InstallRequirement], |
| 39 | install_options: List[str], |
| 40 | global_options: Sequence[str], |
| 41 | root: Optional[str], |
| 42 | home: Optional[str], |
| 43 | prefix: Optional[str], |
| 44 | warn_script_location: bool, |
| 45 | use_user_site: bool, |
| 46 | pycompile: bool, |
| 47 | ) -> List[InstallationResult]: |
| 48 | """ |
| 49 | Install everything in the given list. |
| 50 | |
| 51 | (to be called after having downloaded and unpacked the packages) |
| 52 | """ |
| 53 | to_install = collections.OrderedDict(_validate_requirements(requirements)) |
| 54 | |
| 55 | if to_install: |
| 56 | logger.info( |
| 57 | "Installing collected packages: %s", |
| 58 | ", ".join(to_install.keys()), |
| 59 | ) |
| 60 | |
| 61 | installed = [] |
| 62 | |
| 63 | with indent_log(): |
| 64 | for req_name, requirement in to_install.items(): |
| 65 | if requirement.should_reinstall: |
| 66 | logger.info("Attempting uninstall: %s", req_name) |
| 67 | with indent_log(): |
| 68 | uninstalled_pathset = requirement.uninstall(auto_confirm=True) |
| 69 | else: |
| 70 | uninstalled_pathset = None |
| 71 | |
| 72 | try: |
| 73 | requirement.install( |
| 74 | install_options, |
| 75 | global_options, |
| 76 | root=root, |
| 77 | home=home, |
| 78 | prefix=prefix, |
| 79 | warn_script_location=warn_script_location, |
| 80 | use_user_site=use_user_site, |
| 81 | pycompile=pycompile, |
| 82 | ) |
| 83 | except Exception: |
| 84 | # if install did not succeed, rollback previous uninstall |
| 85 | if uninstalled_pathset and not requirement.install_succeeded: |
| 86 | uninstalled_pathset.rollback() |
| 87 | raise |
| 88 | else: |
| 89 | if uninstalled_pathset and requirement.install_succeeded: |
| 90 | uninstalled_pathset.commit() |
| 91 | |
| 92 | installed.append(InstallationResult(req_name)) |
| 93 | |
| 94 | return installed |