Spaces:
Sleeping
Sleeping
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], | |
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( | |
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 | |