from __future__ import annotations import itertools import os import shlex import shutil import sys from pathlib import Path from typing import cast import click from pip._internal.commands import create_command from pip._internal.commands.install import InstallCommand from pip._internal.index.package_finder import PackageFinder from pip._internal.metadata import get_environment from .. import sync from .._compat import Distribution, parse_requirements from ..exceptions import PipToolsError from ..logging import log from ..repositories import PyPIRepository from ..utils import ( flat_map, get_pip_version_for_python_executable, get_required_pip_specification, get_sys_path_for_python_executable, ) from . import options DEFAULT_REQUIREMENTS_FILE = "requirements.txt" @click.command( name="pip-sync", context_settings={"help_option_names": options.help_option_names} ) @options.version @options.ask @options.dry_run @options.force @options.find_links @options.index_url @options.extra_index_url @options.trusted_host @options.no_index @options.python_executable @options.verbose @options.quiet @options.user @options.cert @options.client_cert @options.src_files @options.pip_args @options.config @options.no_config def cli( ask: bool, dry_run: bool, force: bool, find_links: tuple[str, ...], index_url: str | None, extra_index_url: tuple[str, ...], trusted_host: tuple[str, ...], no_index: bool, python_executable: str | None, verbose: int, quiet: int, user_only: bool, cert: str | None, client_cert: str | None, src_files: tuple[str, ...], pip_args_str: str | None, config: Path | None, no_config: bool, ) -> None: """Synchronize virtual environment with requirements.txt.""" log.verbosity = verbose - quiet if not src_files: if os.path.exists(DEFAULT_REQUIREMENTS_FILE): src_files = (DEFAULT_REQUIREMENTS_FILE,) else: msg = "No requirement files given and no {} found in the current directory" log.error(msg.format(DEFAULT_REQUIREMENTS_FILE)) sys.exit(2) if any(src_file.endswith(".in") for src_file in src_files): msg = ( "Some input files have the .in extension, which is most likely an error " "and can cause weird behaviour. You probably meant to use " "the corresponding *.txt file?" ) if force: log.warning("WARNING: " + msg) else: log.error("ERROR: " + msg) sys.exit(2) if config: log.debug(f"Using pip-tools configuration defaults found in '{config !s}'.") if python_executable: _validate_python_executable(python_executable) install_command = cast(InstallCommand, create_command("install")) options, _ = install_command.parse_args([]) session = install_command._build_session(options) finder = install_command._build_package_finder(options=options, session=session) # Parse requirements file. Note, all options inside requirements file # will be collected by the finder. requirements = flat_map( lambda src: parse_requirements(src, finder=finder, session=session), src_files ) try: merged_requirements = sync.merge(requirements, ignore_conflicts=force) except PipToolsError as e: log.error(str(e)) sys.exit(2) paths = ( None if python_executable is None else get_sys_path_for_python_executable(python_executable) ) installed_dists = _get_installed_distributions( user_only=user_only, local_only=python_executable is None, paths=paths, ) to_install, to_uninstall = sync.diff(merged_requirements, installed_dists) install_flags = _compose_install_flags( finder, no_index=no_index, index_url=index_url, extra_index_url=extra_index_url, trusted_host=trusted_host, find_links=find_links, user_only=user_only, cert=cert, client_cert=client_cert, ) + shlex.split(pip_args_str or "") sys.exit( sync.sync( to_install, to_uninstall, dry_run=dry_run, install_flags=install_flags, ask=ask, python_executable=python_executable, ) ) def _validate_python_executable(python_executable: str) -> None: """ Validates incoming python_executable argument passed to CLI. """ resolved_python_executable = shutil.which(python_executable) if resolved_python_executable is None: msg = "Could not resolve '{}' as valid executable path or alias." log.error(msg.format(python_executable)) sys.exit(2) # Ensure that target python executable has the right version of pip installed pip_version = get_pip_version_for_python_executable(python_executable) required_pip_specification = get_required_pip_specification() if not required_pip_specification.contains(pip_version, prereleases=True): msg = ( "Target python executable '{}' has pip version {} installed. " "Version {} is expected." ) log.error( msg.format(python_executable, pip_version, required_pip_specification) ) sys.exit(2) def _compose_install_flags( finder: PackageFinder, no_index: bool, index_url: str | None, extra_index_url: tuple[str, ...], trusted_host: tuple[str, ...], find_links: tuple[str, ...], user_only: bool, cert: str | None, client_cert: str | None, ) -> list[str]: """ Compose install flags with the given finder and CLI options. """ result = [] # Build --index-url/--extra-index-url/--no-index if no_index: result.append("--no-index") elif index_url is not None: result.extend(["--index-url", index_url]) elif finder.index_urls: finder_index_url = finder.index_urls[0] if finder_index_url != PyPIRepository.DEFAULT_INDEX_URL: result.extend(["--index-url", finder_index_url]) for extra_index in finder.index_urls[1:]: result.extend(["--extra-index-url", extra_index]) else: result.append("--no-index") for extra_index in extra_index_url: result.extend(["--extra-index-url", extra_index]) # Build --trusted-hosts for host in itertools.chain(trusted_host, finder.trusted_hosts): result.extend(["--trusted-host", host]) # Build --find-links for link in itertools.chain(find_links, finder.find_links): result.extend(["--find-links", link]) # Build format controls --no-binary/--only-binary for format_control in ("no_binary", "only_binary"): formats = getattr(finder.format_control, format_control) if not formats: continue result.extend( ["--" + format_control.replace("_", "-"), ",".join(sorted(formats))] ) if user_only: result.append("--user") if cert is not None: result.extend(["--cert", cert]) if client_cert is not None: result.extend(["--client-cert", client_cert]) return result def _get_installed_distributions( local_only: bool = True, user_only: bool = False, paths: list[str] | None = None, ) -> list[Distribution]: """Return a list of installed Distribution objects.""" env = get_environment(paths) dists = env.iter_installed_distributions( local_only=local_only, user_only=user_only, skip=[], ) return [Distribution.from_pip_distribution(dist) for dist in dists]