This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.tool_util.deps.docker_util

"""Utilities for building up Docker commands...

...using common defaults and configuration mechanisms.
import os
import shlex
import sys
from typing import (

    from .container_volumes import DockerVolume

from galaxy.util.commands import argv_to_str

DEFAULT_SET_USER = None if sys.platform == "darwin" else "$UID"

[docs]def kill_command(container: str, signal: Optional[str] = None, **kwds) -> List[str]: args = (["-s", signal] if signal else []) + [container] return command_list("kill", args, **kwds)
[docs]def logs_command(container: str, **kwds) -> List[str]: return command_list("logs", [container], **kwds)
[docs]def build_command(image: str, docker_build_path: str, **kwds) -> List[str]: if os.path.isfile(docker_build_path): docker_build_path = os.path.dirname(os.path.abspath(docker_build_path)) return command_list("build", ["-t", image, docker_build_path], **kwds)
[docs]def build_save_image_command(image: str, destination: str, **kwds) -> List[str]: return command_list("save", ["-o", destination, image], **kwds)
[docs]def build_pull_command(tag: str, **kwds) -> List[str]: return command_list("pull", [tag], **kwds)
[docs]def build_docker_cache_command(image: str, **kwds) -> str: assert kwds.get("to_str", True) inspect_image_command = command_shell("inspect", [image], **kwds) pull_image_command = command_shell("pull", [image], **kwds) cache_command = f"{inspect_image_command} > /dev/null 2>&1\n[ $? -ne 0 ] && {pull_image_command} > /dev/null 2>&1\n" return cache_command
[docs]def build_docker_images_command(truncate=True, **kwds) -> Union[str, List[str]]: args = ["--no-trunc"] if not truncate else [] return command_shell("images", args, **kwds)
[docs]def build_docker_load_command(**kwds) -> Union[str, List[str]]: return command_shell("load", [])
[docs]def build_docker_simple_command( command: str, docker_cmd: str = DEFAULT_DOCKER_COMMAND, sudo: bool = DEFAULT_SUDO, sudo_cmd: str = DEFAULT_SUDO_COMMAND, container_name: Optional[str] = None, **kwd, ) -> str: command_parts = _docker_prefix( docker_cmd=docker_cmd, sudo=sudo, sudo_cmd=sudo_cmd, ) command_parts.append(command) command_parts.append(container_name or "{CONTAINER_NAME}") return " ".join(command_parts)
[docs]def build_docker_run_command( container_command: str, image: str, interactive: bool = False, terminal: bool = False, tag: Optional[str] = None, volumes: Optional[List["DockerVolume"]] = None, volumes_from: Optional[str] = DEFAULT_VOLUMES_FROM, memory: Optional[str] = DEFAULT_MEMORY, env_directives: Optional[List[str]] = None, working_directory: Optional[str] = DEFAULT_WORKING_DIRECTORY, name: Optional[str] = None, net: Optional[str] = DEFAULT_NET, run_extra_arguments: Optional[str] = DEFAULT_RUN_EXTRA_ARGUMENTS, docker_cmd: str = DEFAULT_DOCKER_COMMAND, sudo: bool = DEFAULT_SUDO, sudo_cmd: str = DEFAULT_SUDO_COMMAND, auto_rm: bool = DEFAULT_AUTO_REMOVE, set_user: Optional[str] = DEFAULT_SET_USER, host: Optional[str] = DEFAULT_HOST, guest_ports: Union[bool, List[str]] = False, container_name: Optional[str] = None, ) -> str: env_directives = env_directives or [] volumes = volumes or [] command_parts = _docker_prefix(docker_cmd=docker_cmd, sudo=sudo, sudo_cmd=sudo_cmd, host=host) command_parts.append("run") if interactive: command_parts.append("-i") if terminal: command_parts.append("-t") for env_directive in env_directives: # e.g. -e "GALAXY_SLOTS=$GALAXY_SLOTS" # These are environment variable expansions so we don't quote these. command_parts.extend(["-e", env_directive]) if guest_ports is True: # When is True, expose all ports command_parts.append("-P") elif guest_ports: if not isinstance(guest_ports, list): guest_ports = [guest_ports] for guest_port in guest_ports: command_parts.extend(["-p", guest_port]) if container_name: command_parts.extend(["--name", container_name]) for volume in volumes: command_parts.extend(["-v", str(volume)]) if volumes_from: command_parts.extend(["--volumes-from", shlex.quote(str(volumes_from))]) if memory: command_parts.extend(["-m", shlex.quote(memory)]) command_parts.extend(["--cpus", "${GALAXY_SLOTS:-1}"]) if name: command_parts.extend(["--name", shlex.quote(name)]) if working_directory: command_parts.extend(["-w", shlex.quote(working_directory)]) if net: command_parts.extend(["--net", shlex.quote(net)]) if auto_rm: command_parts.append("--rm") if run_extra_arguments: command_parts.append(run_extra_arguments) if set_user: user = set_user if set_user == DEFAULT_SET_USER: # If future-us is ever in here and fixing this for docker-machine just # use cwltool.docker_id - it takes care of this default nicely. euid = os.geteuid() egid = os.getgid() user = "%d:%d" % (euid, egid) command_parts.extend(["--user", user]) full_image = image if tag: full_image = f"{full_image}:{tag}" command_parts.append(shlex.quote(full_image)) command_parts.append(container_command) return " ".join(command_parts)
[docs]def command_list(command: str, command_args: Optional[List[str]] = None, **kwds) -> List[str]: """Return Docker command as an argv list.""" command_args = command_args or [] command_parts = _docker_prefix(**kwds) command_parts.append(command) command_parts.extend(command_args) return command_parts
[docs]def command_shell(command: str, command_args: Optional[List[str]] = None, **kwds) -> Union[str, List[str]]: """Return Docker command as a string for a shell or command-list.""" command_args = command_args or [] cmd = command_list(command, command_args, **kwds) to_str = kwds.get("to_str", True) if to_str: return argv_to_str(cmd) else: return cmd
def _docker_prefix( docker_cmd: str = DEFAULT_DOCKER_COMMAND, sudo: bool = DEFAULT_SUDO, sudo_cmd: str = DEFAULT_SUDO_COMMAND, host: Optional[str] = DEFAULT_HOST, **kwds, ) -> List[str]: """Prefix to issue a docker command.""" command_parts = [] if sudo: command_parts.append(sudo_cmd) command_parts.append(docker_cmd) if host: command_parts.extend(["-H", host]) return command_parts
[docs]def parse_port_text(port_text): """ >>> slurm_ports = parse_port_text("8888/tcp ->") >>> slurm_ports[8888]['host'] '' >>> ports = parse_port_text("5432/tcp -> :::5432") >>> len(ports) 0 >>> ports_new = parse_port_text("5432/tcp -> [::]:5432") >>> len(ports_new) 0 """ ports = None if port_text is not None: ports = {} for line in port_text.strip().split("\n"): if " -> " not in line: raise Exception(f"Cannot parse host and port from line [{line}]") tool, host = line.split(" -> ", 1) hostname, port = host.rsplit(":", 1) if hostname in ["::", "[::]"]: # Skip unspecified IPv6 address, which is also specified as 0:0:0:0 in another line. # This is brittle of course, but so is parsing the container ports like this. continue port = int(port) tool_p, tool_prot = tool.split("/") tool_p = int(tool_p) ports[tool_p] = dict(tool_port=tool_p, host=hostname, port=port, protocol=tool_prot) return ports