Source code for planemo.conda

"""Planemo specific utilities for dealing with conda.

The extend galaxy-tool-util's features with planemo specific idioms.

import collections
import os
import threading
from copy import deepcopy
from typing import (

from galaxy.tool_util.deps import conda_util
from galaxy.tool_util.deps.conda_util import (
from galaxy.util import unicodify

from planemo.exit_codes import (
from import (
from import yield_tool_sources_on_paths

    from planemo.cli import PlanemoCliContext

BEST_PRACTICE_CHANNELS = ["conda-forge", "bioconda", "defaults"]

[docs] def build_conda_context(ctx: "PlanemoCliContext", **kwds) -> CondaContext: """Build a galaxy-tool-util CondaContext tailored to planemo use. Using planemo's common command-line/global config options. """ condarc_override_default = os.path.join(ctx.workspace, "condarc") conda_prefix = kwds.get("conda_prefix", None) conda_exec = kwds.get("conda_exec", None) use_planemo_shell = kwds.get("use_planemo_shell_exec", True) ensure_channels = kwds.get("conda_ensure_channels", "") condarc_override = kwds.get("condarc", condarc_override_default) use_local = kwds.get("conda_use_local", False) shell_exec = shell if use_planemo_shell else None conda_context = CondaContext( conda_prefix=conda_prefix, ensure_channels=ensure_channels, condarc_override=condarc_override, use_local=use_local, shell_exec=shell_exec, conda_exec=conda_exec, ) handle_auto_init = kwds.get("handle_auto_init", False) if handle_auto_init and not conda_context.is_installed(): auto_init = kwds.get("conda_auto_init", True) failed = True if auto_init: if conda_context.can_install_conda(): if conda_util.install_conda(conda_context) != 0: error("Attempted to install conda and failed.") else: failed = False else: error("Cannot install Conda - perhaps due to a failed installation or permission problems.") else: error("Conda not configured - run ``planemo conda_init`` or pass ``--conda_auto_init`` to continue.") if failed: raise ExitCodeException(EXIT_CODE_FAILED_DEPENDENCIES) if handle_auto_init: if conda_context.ensure_conda_build_installed_if_needed() != 0: error("Attempted to install conda-build and failed.") return conda_context
[docs] def collect_conda_targets(ctx, paths, recursive=False, found_tool_callback=None): """Load CondaTarget objects from supplied artifact sources. If a tool contains more than one requirement, the requirements will each appear once in the output. """ conda_targets = set() real_paths = [] for path in paths: if not os.path.exists(path): targets = target_str_to_targets(path) [conda_targets.add(_) for _ in targets] else: real_paths.append(path) for tool_path, tool_source in yield_tool_sources_on_paths( ctx, real_paths, recursive=recursive, exclude_deprecated=True ): if found_tool_callback: found_tool_callback(tool_path) for target in tool_source_conda_targets(tool_source): conda_targets.add(target) return conda_targets
# Copied and modified from mulled stuff - need to syncronize these concepts. def target_str_to_targets(targets_raw: str) -> List[CondaTarget]: def parse_target(target_str: str) -> CondaTarget: if "=" in target_str: package_name, version = target_str.split("=", 1) else: package_name = target_str version = None target = CondaTarget(package_name, version) return target targets = [parse_target(_) for _ in targets_raw.split(",")] return targets
[docs] def collect_conda_target_lists( ctx: "PlanemoCliContext", paths: Iterable[str], recursive: bool = False, found_tool_callback=None ) -> List[FrozenSet[CondaTarget]]: """Load CondaTarget lists from supplied artifact sources. If a tool contains more than one requirement, the requirements will all appear together as one list element of the output list. """ conda_target_lists, _ = collect_conda_target_lists_and_tool_paths( ctx, paths, recursive=recursive, found_tool_callback=found_tool_callback ) return conda_target_lists
[docs] def collect_conda_target_lists_and_tool_paths( ctx: "PlanemoCliContext", paths: Iterable[str], recursive: bool = False, found_tool_callback=None ) -> Tuple[List[FrozenSet[CondaTarget]], List[List[str]]]: """Load CondaTarget lists from supplied artifact sources. If a tool contains more than one requirement, the requirements will all appear together as one list element of the output list. """ conda_target_sets: Set[FrozenSet[CondaTarget]] = set() tool_paths = collections.defaultdict(list) for tool_path, tool_source in yield_tool_sources_on_paths(ctx, paths, recursive=recursive, yield_load_errors=False): try: if found_tool_callback: found_tool_callback(tool_path) targets = frozenset(tool_source_conda_targets(tool_source)) conda_target_sets.add(targets) tool_paths[targets].append(tool_path) except Exception as e: ctx.log(f"Error while collecting list of conda targets for '{tool_path}': {unicodify(e)}") # Turn them into lists so the order matches before returning... conda_target_lists = list(conda_target_sets) conda_target_tool_paths = [tool_paths[c] for c in conda_target_lists] return conda_target_lists, conda_target_tool_paths
[docs] def tool_source_conda_targets(tool_source): """Load CondaTarget object from supplied abstract tool source.""" requirements, *_ = tool_source.parse_requirements_and_containers() return conda_util.requirements_to_conda_targets(requirements)
best_practice_search_first = threading.local() __all__ = ( "BEST_PRACTICE_CHANNELS", "best_practice_search", "build_conda_context", "collect_conda_targets", "collect_conda_target_lists", "collect_conda_target_lists_and_tool_paths", "tool_source_conda_targets", )