1from __future__ import annotations
2
3import collections
4import logging
5from collections.abc import Generator
6from dataclasses import dataclass
7
8from pip._internal.cli.progress_bars import BarType, get_install_progress_renderer
9from pip._internal.utils.logging import indent_log
10
11from .req_file import parse_requirements
12from .req_install import InstallRequirement
13from .req_set import RequirementSet
14
15__all__ = [
16 "RequirementSet",
17 "InstallRequirement",
18 "parse_requirements",
19 "install_given_reqs",
20]
21
22logger = logging.getLogger(__name__)
23
24
25@dataclass(frozen=True)
26class InstallationResult:
27 name: str
28
29
30def _validate_requirements(
31 requirements: list[InstallRequirement],
32) -> Generator[tuple[str, InstallRequirement], None, None]:
33 for req in requirements:
34 assert req.name, f"invalid to-be-installed requirement: {req}"
35 yield req.name, req
36
37
38def install_given_reqs(
39 requirements: list[InstallRequirement],
40 root: str | None,
41 home: str | None,
42 prefix: str | None,
43 warn_script_location: bool,
44 use_user_site: bool,
45 pycompile: bool,
46 progress_bar: BarType,
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 show_progress = logger.isEnabledFor(logging.INFO) and len(to_install) > 1
64
65 items = iter(to_install.values())
66 if show_progress:
67 renderer = get_install_progress_renderer(
68 bar_type=progress_bar, total=len(to_install)
69 )
70 items = renderer(items)
71
72 with indent_log():
73 for requirement in items:
74 req_name = requirement.name
75 assert req_name is not None
76 if requirement.should_reinstall:
77 logger.info("Attempting uninstall: %s", req_name)
78 with indent_log():
79 uninstalled_pathset = requirement.uninstall(auto_confirm=True)
80 else:
81 uninstalled_pathset = None
82
83 try:
84 requirement.install(
85 root=root,
86 home=home,
87 prefix=prefix,
88 warn_script_location=warn_script_location,
89 use_user_site=use_user_site,
90 pycompile=pycompile,
91 )
92 except Exception:
93 # if install did not succeed, rollback previous uninstall
94 if uninstalled_pathset and not requirement.install_succeeded:
95 uninstalled_pathset.rollback()
96 raise
97 else:
98 if uninstalled_pathset and requirement.install_succeeded:
99 uninstalled_pathset.commit()
100
101 installed.append(InstallationResult(req_name))
102
103 return installed