blob: 8d56359666869463527be82a7de2e7a04eec165b [file] [log] [blame]
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