Source code for planemo.commands.cmd_container_register

"""Module describing the planemo ``container_register`` command."""
import os
import string

import click
from galaxy.tool_util.deps.container_resolvers.mulled import targets_to_mulled_name
from galaxy.tool_util.deps.mulled.mulled_build import (
    base_image_for_targets,
    DEFAULT_BASE_IMAGE,
)
from galaxy.tool_util.deps.mulled.util import (
    conda_build_target_str,
    v2_image_name,
)

from planemo import options
from planemo.cli import command_function
from planemo.conda import (
    best_practice_search,
    build_conda_context,
    collect_conda_target_lists_and_tool_paths,
)
from planemo.git import (
    add,
    branch,
    commit,
    push,
)
from planemo.github_util import (
    clone_fork_branch,
    DEFAULT_REMOTE_NAME,
    get_repository_object,
    pull_request,
)
from planemo.mulled import conda_to_mulled_targets

REGISTRY_TARGET_NAME = "multi-package-containers"
REGISTRY_TARGET_PATH = "combinations"
REGISTRY_REPOSITORY = "BioContainers/multi-package-containers"
DEFAULT_MESSAGE = "Add container $hash.\n**Hash**: $hash\n\n**Packages**:\n$packages\nBase Image:$base_image\n\n"
DEFAULT_MESSAGE += "**For** :\n$tools\n\nGenerated with Planemo."
CONTENTS = "#targets\tbase_image\timage_build\n$targets\t$base_image\t$image_build\n"
BIOCONTAINERS_PLATFORM = "linux-64"


@click.command("container_register")
@options.optional_tools_arg(multiple=True)
@options.recursive_option()
@options.mulled_namespace_option()
@options.conda_target_options()
@click.option(
    "output_directory",
    "--output_directory",
    type=click.Path(
        file_okay=False,
        dir_okay=True,
        resolve_path=True,
    ),
    default=None,
    help=("Container registration directory (defaults to ~/.planemo/multi-package-containers."),
)
@click.option(
    "-m",
    "--message",
    default=DEFAULT_MESSAGE,
    help="Commit and pull request message template for registration interactions.",
)
@click.option(
    "--pull_request/--no_pull_request",
    is_flag=True,
    default=True,
    help="Fork and create a pull request against %s for these changes." % REGISTRY_REPOSITORY,
)
@click.option(
    "--force_push/--no_force_push",
    is_flag=True,
    default=False,
    help="Force push branch for pull request in case it already exists.",
)
@command_function
def cli(ctx, paths, **kwds):
    """Register multi-requirement containers as needed.

    BioContainers publishes all Bioconda packages automatically as individual
    container images. These however are not enough for tools with multiple
    best-practice requirements. Such requirements should be recorded and published
    so that a container can be created and registered for these tools.
    """
    registry_target = RegistryTarget(ctx, **kwds)
    conda_context = build_conda_context(ctx, **kwds)

    combinations_added = 0
    conda_targets_list, tool_paths_list = collect_conda_target_lists_and_tool_paths(
        ctx, paths, recursive=kwds["recursive"]
    )
    for conda_targets, tool_paths in zip(conda_targets_list, tool_paths_list):
        ctx.vlog("Handling conda_targets [%s]" % conda_targets)
        mulled_targets = conda_to_mulled_targets(conda_targets)
        mulled_targets_str = "- " + "\n- ".join(map(conda_build_target_str, mulled_targets))

        if len(mulled_targets) < 1:
            ctx.log("Skipping registration, no targets discovered.")
            continue

        name = v2_image_name(mulled_targets)
        tag = "0"
        name_and_tag = f"{name}-{tag}"
        target_filename = os.path.join(registry_target.output_directory, "%s.tsv" % name_and_tag)
        if os.path.exists(target_filename):
            ctx.log("Target file '%s' already exists, skipping" % target_filename)
            continue

        if targets_to_mulled_name(mulled_targets, hash_func="v2", namespace=kwds["mulled_namespace"]):
            ctx.vlog("quay repository already exists, skipping")
            continue

        if registry_target.has_pull_request_for(name):
            ctx.vlog("Found matching open pull request for [%s], skipping" % name)
            continue

        best_practice_requirements = True
        for conda_target in conda_targets:
            best_hit, exact = best_practice_search(
                conda_target, conda_context=conda_context, platform=BIOCONTAINERS_PLATFORM
            )
            if not best_hit:
                ctx.log("Target [%s] is not available in best practice channels - skipping" % conda_target)
                best_practice_requirements = False
            if not exact:
                ctx.log(
                    "Target version [%s] for package [%s] is not available in best practice channels - please specify the full version",
                    conda_target.version,
                    conda_target.package,
                )

        if not best_practice_requirements:
            continue

        base_image = DEFAULT_BASE_IMAGE
        for conda_target in conda_targets:
            base_image = base_image_for_targets([conda_target], conda_context=conda_context)
            if base_image != DEFAULT_BASE_IMAGE:
                ctx.log(f"{conda_target} requires '{base_image}' as base image")
                break

        registry_target.write_targets(ctx, target_filename, mulled_targets, tag, base_image)
        tools_str = "\n".join(map(lambda p: "- " + os.path.basename(p), tool_paths))
        registry_target.handle_pull_request(
            ctx, name, target_filename, mulled_targets_str, tools_str, base_image, **kwds
        )
        combinations_added += 1


[docs]class RegistryTarget: """Abstraction around mulled container registry (both directory and Github repo).""" def __init__(self, ctx, **kwds): output_directory = kwds["output_directory"] pr_titles = [] target_repository = None self.remote_name = DEFAULT_REMOTE_NAME do_pull_request = kwds.get("pull_request", True) if output_directory is None: target_repository = os.path.join(ctx.workspace, REGISTRY_TARGET_NAME) output_directory = os.path.join(target_repository, REGISTRY_TARGET_PATH) self.remote_name = ( clone_fork_branch( ctx, "https://github.com/%s" % REGISTRY_REPOSITORY, target_repository, fork=do_pull_request, ) or self.remote_name ) pr_titles = [pr.title for pr in open_prs(ctx)] self.do_pull_request = do_pull_request self.pr_titles = pr_titles self.output_directory = output_directory self.target_repository = target_repository
[docs] def has_pull_request_for(self, name): has_pr = False if self.do_pull_request: if any([name in t for t in self.pr_titles]): has_pr = True return has_pr
[docs] def handle_pull_request(self, ctx, name, target_filename, packages_str, tools_str, base_image, **kwds): if self.do_pull_request: message = kwds["message"] message = string.Template(message).safe_substitute( { "hash": name, "packages": packages_str, "tools": tools_str, "base_image": base_image, } ) branch_name = name.replace(":", "-") branch(ctx, self.target_repository, branch_name, from_branch="master") add(ctx, self.target_repository, target_filename) commit(ctx, self.target_repository, message=message) force_push = kwds.get("force_push", False) push(ctx, repo_path=self.target_repository, to=self.remote_name, branch=branch_name, force=force_push) pull_request(ctx, self.target_repository, message=message, repo=REGISTRY_REPOSITORY)
[docs] def write_targets(self, ctx, target_filename, mulled_targets, tag, base_image): with open(target_filename, "w") as f: targets = to_target_str(mulled_targets) f.write(string.Template(CONTENTS).safe_substitute(targets=targets, base_image=base_image, image_build=tag)) ctx.log(f"Wrote requirements [{targets}] to file [{target_filename}]")
[docs]def to_target_str(targets): target_strings = [] for target in targets: if target.version: target_str = f"{target.package_name}={target.version}" else: target_str = target.package_name target_strings.append(target_str) return ",".join(target_strings)
[docs]def open_prs(ctx): repo = get_repository_object(ctx, REGISTRY_REPOSITORY) prs = [pr for pr in repo.get_pulls()] return prs