Quelle  sync.py   Sprache: Python

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]