Source code for galaxy.tool_util.deps.conda_util

import functools
import hashlib
import json
import logging
import os
import re
import shutil
import sys
import tempfile
from typing import (

import packaging.version

from galaxy.util import (
from . import installable

log = logging.getLogger(__name__)

# Not sure there are security concerns, lets just fail fast if we are going
# break shell commands we are building.
SHELL_UNSAFE_PATTERN = re.compile(r"[\s\"']")

IS_OS_X = sys.platform == "darwin"

VERSIONED_ENV_DIR_NAME = re.compile(r"__(.*)@(.*)")
UNVERSIONED_ENV_DIR_NAME = re.compile(r"__(.*)@_uv_")
CONDA_PACKAGE_SPECS = ("conda=4.6.14", "'pyopenssl>=22.1.0'")

def conda_link():
    if IS_OS_X:
        url = ""
        if sys.maxsize > 2**32:
            url = ""
            url = ""
    return url

def find_conda_prefix(conda_prefix=None):
    """If supplied conda_prefix is not set, default to the default location
    for Miniconda installs.
    if conda_prefix is None:
        home = os.path.expanduser("~")
        miniconda_2_dest = os.path.join(home, "miniconda2")
        miniconda_3_dest = os.path.join(home, "miniconda3")
        anaconda_2_dest = os.path.join(home, "anaconda2")
        anaconda_3_dest = os.path.join(home, "anaconda3")
        # Prefer miniconda3 install if both available
        if os.path.exists(miniconda_3_dest):
            return miniconda_3_dest
        elif os.path.exists(miniconda_2_dest):
            return miniconda_2_dest
        elif os.path.exists(anaconda_3_dest):
            return anaconda_3_dest
        elif os.path.exists(anaconda_2_dest):
            return anaconda_2_dest
            return miniconda_3_dest
    return conda_prefix

[docs]class CondaContext(installable.InstallableContext): installable_description = "Conda"
[docs] def __init__( self, conda_prefix=None, conda_exec=None, shell_exec=None, debug=False, ensure_channels="", condarc_override=None, use_path_exec=USE_PATH_EXEC_DEFAULT, copy_dependencies=False, use_local=USE_LOCAL_DEFAULT, ): self.condarc_override = condarc_override if not conda_exec and use_path_exec: conda_exec = which("conda") if conda_exec and isinstance(conda_exec, str): conda_exec = os.path.normpath(conda_exec) self.conda_exec = conda_exec self.debug = debug self.shell_exec = shell_exec or self.copy_dependencies = copy_dependencies if conda_prefix is None: info = self.conda_info() if info and "default_prefix" in info: conda_prefix = info["default_prefix"] if conda_prefix is None: conda_prefix = find_conda_prefix(conda_prefix) self.conda_prefix = conda_prefix if conda_exec is None: self.conda_exec = self._bin("conda") self.ensure_channels: List[str] = listify(ensure_channels) self.use_local = use_local self._reset_conda_properties()
def _reset_conda_properties(self): self._conda_version = None self._conda_build_available = None @property def conda_version(self): if self._conda_version is None: self._guess_conda_properties() return self._conda_version @property def conda_build_available(self): if self._conda_build_available is None: self._guess_conda_properties() return self._conda_build_available def _guess_conda_properties(self): info = self.conda_info() self._conda_version = packaging.version.parse(info["conda_version"]) self._conda_build_available = False conda_build_version = info.get("conda_build_version") if conda_build_version and conda_build_version != "not installed": try: packaging.version.parse(conda_build_version) self._conda_build_available = True except Exception: pass @property def _override_channels_args(self): override_channels_args = [] if self.ensure_channels: override_channels_args.append("--override-channels") for channel in self.ensure_channels: override_channels_args.extend(["--channel", channel]) return override_channels_args
[docs] def ensure_conda_build_installed_if_needed(self): if self.use_local and not self.conda_build_available: conda_targets = [CondaTarget("conda-build", version=CONDA_BUILD_VERSION)] # Cannot use --use-local during installation of conda-build. return install_conda_targets(conda_targets, conda_context=self, env_name=None, allow_local=False) else: return 0
[docs] def conda_info(self): if self.conda_exec is not None: cmd = listify(self.conda_exec) + ["info", "--json"] info_out = commands.execute(cmd) info = json.loads(info_out) return info else: return None
[docs] def is_conda_installed(self): """ Check if conda_exec exists """ if os.path.exists(self.conda_exec): return True else: return False
[docs] def can_install_conda(self): """ If conda_exec is set to a path outside of conda_prefix, there is no use installing conda into conda_prefix, since it can't be used by galaxy. If conda_exec equals conda_prefix/bin/conda, we can install conda if either conda_prefix does not exist or is empty. """ conda_exec = os.path.abspath(self.conda_exec) conda_prefix_plus_exec = os.path.abspath(os.path.join(self.conda_prefix, "bin/conda")) if conda_exec == conda_prefix_plus_exec: if not os.path.exists(self.conda_prefix): return True elif os.listdir(self.conda_prefix) == []: os.rmdir(self.conda_prefix) # Conda's install script fails if path exists (even if empty). return True else: log.warning( "Cannot install Conda because conda_prefix '%s' exists and is not empty.", self.conda_prefix ) return False else: log.warning( "Skipping installation of Conda into conda_prefix '%s', " "since conda_exec '%s' is set to a path outside of conda_prefix.", self.conda_prefix, self.conda_exec, ) return False
[docs] def exec_command(self, operation, args, stdout_path=None): """ Execute the requested command. Return the process exit code (i.e. 0 in case of success). """ cmd = listify(self.conda_exec) + operation.split() if self.debug: cmd.append("--debug") cmd.extend(args) env = {} if self.condarc_override: env["CONDARC"] = self.condarc_override cmd_string = shlex_join(cmd) kwds = dict() try: if stdout_path: kwds["stdout"] = open(stdout_path, "w") cmd_string += f" > '{stdout_path}'" conda_exec_home = env["HOME"] = tempfile.mkdtemp( prefix="conda_exec_home_" ) # We don't want to pollute ~/.conda, which may not even be writable log.debug("Executing command: %s", cmd_string) return self.shell_exec(cmd, env=env, **kwds) except Exception: log.exception("Failed to execute command: %s", cmd_string) return 1 finally: if kwds.get("stdout"): kwds["stdout"].close() if conda_exec_home: shutil.rmtree(conda_exec_home, ignore_errors=True)
[docs] def exec_create(self, args, allow_local=True, stdout_path=None): """ Return the process exit code (i.e. 0 in case of success). """ for try_strict in [True, False]: create_args = ["-y", "--quiet"] if try_strict: if self.conda_version >= packaging.version.parse("4.7.5"): create_args.append("--strict-channel-priority") else: continue if allow_local and self.use_local: create_args.append("--use-local") create_args.extend(self._override_channels_args) create_args.extend(args) ret = self.exec_command("create", create_args, stdout_path=stdout_path) if ret == 0: break return ret
[docs] def exec_remove(self, args): """ Remove a conda environment using conda env remove -y --name `args`. Return the process exit code (i.e. 0 in case of success). """ remove_args = ["-y", "--name"] remove_args.extend(args) return self.exec_command("env remove", remove_args)
[docs] def exec_install(self, args, allow_local=True, stdout_path=None): """ Return the process exit code (i.e. 0 in case of success). """ for try_strict in [True, False]: install_args = ["-y"] if try_strict: if self.conda_version >= packaging.version.parse("4.7.5"): install_args.append("--strict-channel-priority") else: continue if allow_local and self.use_local: install_args.append("--use-local") install_args.extend(self._override_channels_args) install_args.extend(args) ret = self.exec_command("install", install_args, stdout_path=stdout_path) if ret == 0: break if ret == 0: self._reset_conda_properties() return ret
[docs] def exec_clean(self, args=None, quiet=False): """ Clean up after conda installation. Return the process exit code (i.e. 0 in case of success). """ clean_args = ["--tarballs", "-y"] if args: clean_args.extend(args) stdout_path = None if quiet: stdout_path = "/dev/null" return self.exec_command("clean", clean_args, stdout_path=stdout_path)
[docs] def export_list(self, name, path): """ Return the process exit code (i.e. 0 in case of success). """ return self.exec_command("list", ["--name", name, "--export"], stdout_path=path)
[docs] def env_path(self, env_name): return os.path.join(self.envs_path, env_name)
@property def envs_path(self): return os.path.join(self.conda_prefix, "envs")
[docs] def has_env(self, env_name): env_path = self.env_path(env_name) return os.path.isdir(env_path)
@property def deactivate(self): return self._bin("deactivate") @property def activate(self): return self._bin("activate")
[docs] def is_installed(self): return self.is_conda_installed()
[docs] def can_install(self): return self.can_install_conda()
@property def parent_path(self): return os.path.dirname(os.path.abspath(self.conda_prefix)) def _bin(self, name): return os.path.join(self.conda_prefix, "bin", name)
def installed_conda_targets(conda_context): envs_path = conda_context.envs_path dir_contents = os.listdir(envs_path) if os.path.exists(envs_path) else [] for name in dir_contents: versioned_match = VERSIONED_ENV_DIR_NAME.match(name) if versioned_match: yield CondaTarget(, unversioned_match = UNVERSIONED_ENV_DIR_NAME.match(name) if unversioned_match: yield CondaTarget(
[docs]class CondaTarget:
[docs] def __init__(self, package, version=None, channel=None): if is not None: raise ValueError(f"Invalid package [{package}] encountered.") self.package = package if version and is not None: raise ValueError(f"Invalid version [{version}] encountered.") self.version = version if channel and is not None: raise ValueError(f"Invalid version [{channel}] encountered.") = channel
def __str__(self): attributes = f"package={self.package}" if self.version is not None: attributes = f"{self.package},version={self.version}" else: attributes = f"{self.package},unversioned" if attributes = "%s,channel=%s" % return f"CondaTarget[{attributes}]" __repr__ = __str__ @property def package_specifier(self): """Return a package specifier as consumed by conda install/create.""" if self.version: return f"{self.package}={self.version}" else: return self.package @property def install_environment(self): """The dependency resolution and installation frameworks will expect each target to be installed it its own environment with a fixed and predictable name given package and version. """ if self.version: return f"__{self.package}@{self.version}" else: return f"__{self.package}@_uv_" def __hash__(self): return hash((self.package, self.version, def __eq__(self, other): if isinstance(other, self.__class__): return (self.package, self.version, == (other.package, other.version, return False def __ne__(self, other): return not (self == other)
def hash_conda_packages(conda_packages, conda_target=None): """Produce a unique hash on supplied packages. TODO: Ideally we would do this in such a way that preserved environments. """ h ="sha256") for conda_package in conda_packages: h.update(smart_str(conda_package.install_environment)) return h.hexdigest() # shell makes sense for planemo, in Galaxy this should just execute # these commands as Python
[docs]def install_conda(conda_context, force_conda_build=False): with tempfile.NamedTemporaryFile(suffix=".sh", prefix="conda_install", delete=False) as temp: script_path = download_cmd = commands.download_command(conda_link(), to=script_path) install_cmd = ["bash", script_path, "-b", "-p", conda_context.conda_prefix] package_targets = list(CONDA_PACKAGE_SPECS) if force_conda_build or conda_context.use_local: package_targets.append(f"conda-build={CONDA_BUILD_VERSION}")"Installing conda, this may take several minutes.") try: exit_code = conda_context.shell_exec(download_cmd) if exit_code: return exit_code exit_code = conda_context.shell_exec(install_cmd) except Exception: log.exception("Failed to install conda") return 1 finally: if os.path.exists(script_path): os.remove(script_path) if exit_code: return exit_code return conda_context.exec_install(package_targets, allow_local=False)
def install_conda_targets(conda_targets, conda_context, env_name=None, allow_local=True): """ Return the process exit code (i.e. 0 in case of success). """ if env_name is not None: create_args = [ "--name", env_name, # environment for package ] for conda_target in conda_targets: create_args.append(conda_target.package_specifier) return conda_context.exec_create(create_args, allow_local=allow_local) else: return conda_context.exec_install([t.package_specifier for t in conda_targets], allow_local=allow_local)
[docs]def install_conda_target(conda_target, conda_context, skip_environment=False): """ Install specified target into a its own environment. Return the process exit code (i.e. 0 in case of success). """ if not skip_environment: create_args = [ "--name", conda_target.install_environment, # environment for package conda_target.package_specifier, ] return conda_context.exec_create(create_args) else: return conda_context.exec_install([conda_target.package_specifier])
def cleanup_failed_install_of_environment(env, conda_context): if conda_context.has_env(env): conda_context.exec_remove([env]) def cleanup_failed_install(conda_target, conda_context=None): cleanup_failed_install_of_environment(conda_target.install_environment, conda_context=conda_context) def best_search_result( conda_target, conda_context: CondaContext, offline: bool = False, platform: Optional[str] = None ): """Find best "conda search" result for specified target. Return ``None`` if no results match. """ search_args = [conda_target.package] try: res = conda_context.exec_search(search_args, json=True, offline=offline, platform=platform) # Use python's stable list sorting to sort by date, # then build_number, then version. The top of the list # then is the newest version with the newest build and # the latest update time. hits = json.loads(res).get(conda_target.package, [])[::-1] hits = sorted(hits, key=lambda hit: hit["build_number"], reverse=True) hits = sorted(hits, key=lambda hit: packaging.version.parse(hit["version"]), reverse=True) except commands.CommandLineException as e: log.error(f"Could not execute: '{e.command}'\n{e}") hits = [] if len(hits) == 0: return (None, None) best_result = (hits[0], False) for hit in hits: if is_search_hit_exact(conda_target, hit): best_result = (hit, True) break return best_result def is_search_hit_exact(conda_target, search_hit): target_version = conda_target.version # It'd be nice to make request verson of 1.0 match available # version of 1.0.3 or something like that. return not target_version or search_hit["version"] == target_version def is_conda_target_installed(conda_target, conda_context): # fail by default if conda_context.has_env(conda_target.install_environment): return True else: return False def filter_installed_targets(conda_targets, conda_context): installed = functools.partial(is_conda_target_installed, conda_context=conda_context) return list(filter(installed, conda_targets)) def build_isolated_environment( conda_packages, conda_context, path=None, copy=False, quiet=False, ): """Build a new environment (or reuse an existing one from hashes) for specified conda packages. """ if not isinstance(conda_packages, list): conda_packages = [conda_packages] # Lots we could do in here, hashing, checking revisions, etc... tempdir = None try: hash = hash_conda_packages(conda_packages) tempdir = tempfile.mkdtemp(prefix="jobdeps", suffix=hash) tempdir_name = os.path.basename(tempdir) export_paths = [] for conda_package in conda_packages: name = conda_package.install_environment export_path = os.path.join(tempdir, name) conda_context.export_list(name, export_path) export_paths.append(export_path) create_args = ["--unknown"] # Works in 3.19, 4.0 - 4.2 - not in 4.3. # Adjust fix if they fix Conda - xref # - # - offline_works = (conda_context.conda_version < packaging.version.parse("4.3")) or ( conda_context.conda_version >= packaging.version.parse("4.4") ) if offline_works: create_args.extend(["--offline"]) else: create_args.extend(["--use-index-cache"]) if path is None: create_args.extend(["--name", tempdir_name]) else: create_args.extend(["--prefix", path]) if copy: create_args.append("--copy") for export_path in export_paths: create_args.extend(["--file", export_path]) stdout_path = None if quiet: stdout_path = "/dev/null" if path is not None and os.path.exists(path): exit_code = conda_context.exec_install(create_args, stdout_path=stdout_path) else: exit_code = conda_context.exec_create(create_args, stdout_path=stdout_path) return (path or tempdir_name, exit_code) finally: conda_context.exec_clean(quiet=quiet) if tempdir is not None: shutil.rmtree(tempdir) def requirement_to_conda_targets(requirement): conda_target = None if requirement.type == "package": conda_target = CondaTarget(, version=requirement.version) return conda_target
[docs]def requirements_to_conda_targets(requirements): conda_targets = (requirement_to_conda_targets(_) for _ in requirements) return [c for c in conda_targets if c is not None]
__all__ = ( "CondaContext", "CondaTarget", "install_conda", "install_conda_target", "requirements_to_conda_targets", )