fabric/modules/pip.py

import sys
import os
import logging

from fabric.api import env, task
from fabric.operations import run
from fabric.context_managers import hide, settings
from fabric.utils import abort

from .utils import virtualenv_source, virtualenv
from .utils import printerr
from .utils import booleanize, handle_flag_message
from .utils import prompt_continue
from .utils import upload_template

from .initialize import get_config


ERROR_BAD_BRANCH_PARAM = -3
ERROR_BAD_PARAM = -2


@task
def setup_virtualenv(python3=True):
    """
    Sets up a virtual environment using virtualenvwrapper.

    This method creates a new virtual environment with the specified Python
    version. It uses the 'mkvirtualenv' command from virtualenvwrapper and can
    optionally use Python 3.

    :param python3: Specifies whether to use Python 3 for the virtual
                    environment, defaults to True.

    :type python3: bool

    Note:
        - The virtual environment's name is taken from
            `configuration.virtualenv.name`.
        - If `env.debug` is True, the method logs additional debug information.
    """
    configuration = env.config

    if env.debug:
        logging.basicConfig(
            format='\n%(levelname)s: deploy.setup_virtualenv %(message)s',
            level=logging.DEBUG)

    # mkvirtualenv_cmd = "mkvirtualenv --no-site-packages "
    mkvirtualenv_cmd = "mkvirtualenv "

    if python3:
        mkvirtualenv_cmd += "--python=python3 "

    mkvirtualenv_cmd += f"{configuration.virtualenv.name}"

    if env.debug:

        logging.debug("python3 install: %s", python3)

        logging.debug("virtualenv.workon        : %s",
                      configuration.virtualenv.workon)

        logging.debug("virtualenv.activate      : %s",
                      configuration.virtualenv.activate)

        logging.debug("virtualenv.name          : %s",
                      configuration.virtualenv.name)

        logging.debug("virtualenv.paths.bin     : %s",
                      configuration.virtualenv.paths.bin)

        logging.debug("virtualenv.paths.root    : %s",
                      configuration.virtualenv.paths.root)

        logging.debug("mkvirtualenv_cmd: %s", mkvirtualenv_cmd)

        logging.debug(
            "with virtualenv_source(): "
            "run(\"\n\t%s\n\t\")", mkvirtualenv_cmd)
    else:

        # run("source virtualenvwrapper.sh; mkvirtualenv "
        #     "--no-site-packages {virtualenv_name}".format(
        #         virtualenv_name=configuration.virtualenv.name))

        with virtualenv_source():
            run(mkvirtualenv_cmd)


@task
def bootstrap_pip():
    """
    Bootstraps pip in the current environment.

    This method is a wrapper that calls the 'setup' method, intended for
    setting up or bootstrapping pip in the current virtual environment.

    Note:
        - This method currently calls 'setup', which handles the installation
          of packages via pip based on the requirements file.
    """
    # upgrade()
    setup()


@task
def update_pip():
    """
    Updates pip to the latest version.

    This method updates pip, the Python package installer, to its latest
    version within the current virtual environment.

    Note:
        - The method assumes it's running inside a virtual environment.
        - If `env.debug` is True, it logs the upgrade command being executed.
    """

    # configuration = env.config

    if env.debug:
        logging.basicConfig(
            format='\n%(levelname)s: deploy.pip %(message)s',
            level=logging.DEBUG)

    pipinstall_cmd = "pip install --upgrade pip"

    if env.debug:
        logging.debug("with virtualenv(): run(\"\n\t%s\n\t\")",
                      pipinstall_cmd)
    else:
        with virtualenv():
            run(pipinstall_cmd)


@task
def setup():
    """
    Installs all required packages via pip.

    This method installs packages specified in a requirements file using pip.
    The requirements file path is retrieved from
    `configuration.virtualenv.requirements.filepath`.

    Note:
        - Assumes execution within a virtual environment.
        - If `env.debug` is enabled, debug information is logged.
    """
    configuration = env.config

    if env.debug:
        logging.basicConfig(
            format='\n%(levelname)s: deploy.pip %(message)s',
            level=logging.DEBUG)

    pipinstall_cmd = (
        f"pip install -r "
        f"{configuration.virtualenv.requirements.filepath}"
    )

    if env.debug:
        logging.debug("with virtualenv(): run(\"\n\t%s\n\t\")",
                      pipinstall_cmd)
    else:
        with virtualenv():
            run(pipinstall_cmd)


@task
def install(package_name=None):
    """
    Installs a package via pip within a virtual environment.

    This method installs a specified package using pip. If the package_name
    is '--all', it installs all packages listed in the requirements file.

    :param package_name: Name of the package to install or '--all' to install
                         all packages from requirements file, defaults to None.
    :type package_name: str, optional

    :raises ValueError: If no package_name is provided.
    """
    configuration = env.config

    if not package_name:
        abort("You must specify a package name to be installed")

    if package_name == "--all":
        pipinstall_cmd = (
            f"pip install -r "
            f"{configuration.virtualenv.requirements.filepath}"
        )

    else:
        pipinstall_cmd = f"pip install {package_name}"

    if env.debug:
        print(f"pipinstall_cmd: {pipinstall_cmd}")

    else:
        with virtualenv():
            run(pipinstall_cmd)


@task
def uninstall(package_name, silent=False):
    """
    Uninstalls a specified package from the virtual environment.

    This method activates the virtual environment and uses pip to uninstall the
    specified package. It can optionally suppress command line output.

    :param package_name: Name of the package to uninstall.
    :type package_name: str

    :param silent: Suppress command line output if True, defaults to False.
    :type silent: bool
    :return: None
    """
    with virtualenv():
        with settings(
                hide('warnings', 'running', 'stdout', 'stderr'),
                warn_only=True):

            run(f"pip uninstall -y {package_name}", quiet=silent)


@task
def freeze(param='help'):
    """
    True - update package list and freeze output
    """
    configuration = env.config

    msg_help = """
    pip.freeze takes one of three values:
    \thelp - this help message
    \tTrue - update the pip package list the freeze output
    \tFalse (default) - print the freeze output to the console
    """

    if handle_flag_message(param, msg_help, 'help'):
        sys.exit()
    else:
        try:
            param = booleanize(param)
        except TypeError:
            printerr(
                f"the parameter value you gave, '{param}',"
                f" is not a valid parameter\n{msg_help}",
                ERROR_BAD_PARAM
            )

    if param:
        cmd_pipfreeze = (
            f"pip freeze > "
            f"{configuration.virtualenv.requirements.filepath}"
        )

    else:
        cmd_pipfreeze = "pip freeze"

    with virtualenv():
        run(cmd_pipfreeze)


@task
def copyto(branch=None):
    """
    copy requirements from the current branch to the specified branch

    this only changes the requirements on the local branches.  It does not
    upload remotely. Use deploy.sync to perform remote updates.

    Keyword Arguments:
        branch -- the branch to copy to

    Usage:
        fab copyto --branch=staging
    """

    if branch is None:
        printerr(
            "Missing required parameter: 'branch'\n"
            "Usage: fab pip.copyto:<branch>\n"
            "Valid options: development, staging, production",
            ERROR_BAD_BRANCH_PARAM
        )
        return

    configuration = env.config
    branch_list = ['staging', 'production', 'development']

    if branch not in branch_list:
        printerr(
            f"Branch parameter '{branch}' must be one of {branch_list} "
            "values.", ERROR_BAD_BRANCH_PARAM)

    elif branch == 'development':
        printerr(
            "This method does not allow copying to development branch",
            ERROR_BAD_BRANCH_PARAM)

    elif configuration.project.branch == branch:
        printerr(
            "You have set the source branch to the current branch."
            "This will simply copy over \n\tthe requirements file for "
            "this branch with itself", ERROR_BAD_BRANCH_PARAM)

    branch_config = get_config(branch)

    current_local_path = os.path.join(
        configuration.virtualenv.requirements.local,
        configuration.virtualenv.requirements.filename)

    branch_local_path = os.path.join(
        configuration.virtualenv.requirements.local,
        branch_config.virtualenv.requirements.filename)

    print(f"current_local_path: {current_local_path}")
    print("branch_local_path: {branch_local_path}")

    message = (
        f"Copying file from current branch '{configuration.project.branch}'"
        f"to destination branch '{branch}'. Continue? Y/n "
    )

    prompt_continue(message=message)

    upload_template(
        filename=configuration.virtualenv.requirements.filename,
        destination=branch_local_path,
        context=None,
        use_jinja=False,
        use_sudo=False,
        backup=True,
        template_dir=configuration.virtualenv.requirements.local)


def check_package_installed(package_name, silent=False):
    """
    Checks if a specified package is installed in the virtual environment.

    Activates the virtual environment and uses 'pip list' to check if the
    specified package is installed. Can optionally suppress command line
    output.

    :param package_name: Name of the package to check.
    :type package_name: str

    :param silent: Suppress command line output if True, defaults to False.
    :type silent: bool

    :return: True if installed, False otherwise.
    :rtype: bool
    """
    with virtualenv():

        with settings(
                hide('warnings', 'running', 'stdout', 'stderr'),
                warn_only=True):

            output = run(f"pip list | grep {package_name}", quiet=silent)

        return package_name in output