Warning

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.tools.deps.commands

"""Generic I/O and shell processing code used by Galaxy tool dependencies."""
import logging
import os
import subprocess
import sys as _sys

import six
from six.moves import shlex_quote

from galaxy.util import which

log = logging.getLogger(__name__)

STDOUT_INDICATOR = "-"


[docs]def redirecting_io(sys=_sys): """Predicate to determine if we are redicting I/O in process.""" assert sys is not None return not hasattr(sys.stdout, "fileno")
[docs]def redirect_aware_commmunicate(p, sys=_sys): """Variant of process.communicate that works with in process I/O redirection.""" assert sys is not None out, err = p.communicate() if redirecting_io(sys=sys): if out: sys.stdout.write(out) out = None if err: sys.stderr.write(err) err = None return out, err
[docs]def shell(cmds, env=None, **kwds): """Run shell commands with `shell_process` and wait.""" sys = kwds.get("sys", _sys) assert sys is not None p = shell_process(cmds, env, **kwds) if redirecting_io(sys=sys): redirect_aware_commmunicate(p, sys=sys) exit = p.returncode return exit else: return p.wait()
[docs]def shell_process(cmds, env=None, **kwds): """A high-level method wrapping subprocess.Popen. Handles details such as environment extension and in process I/O redirection. """ sys = kwds.get("sys", _sys) popen_kwds = dict() if isinstance(cmds, six.string_types): log.warning("Passing program arguments as a string may be a security hazard if combined with untrusted input") popen_kwds['shell'] = True if kwds.get("stdout", None) is None and redirecting_io(sys=sys): popen_kwds["stdout"] = subprocess.PIPE if kwds.get("stderr", None) is None and redirecting_io(sys=sys): popen_kwds["stderr"] = subprocess.PIPE popen_kwds.update(**kwds) if env: new_env = os.environ.copy() new_env.update(env) popen_kwds["env"] = new_env p = subprocess.Popen(cmds, **popen_kwds) return p
[docs]def execute(cmds): """Execute commands and throw an exception on a non-zero exit. Return the standard output if the commands are successful """ return _wait(cmds, shell=False, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
[docs]def argv_to_str(command_argv, quote=True): """Convert an argv command list to a string for shell subprocess. If None appears in the command list it is simply excluded. Arguments are quoted with shlex_quote. That said, this method is not meant to be used in security critical paths of code and should not be used to sanitize code. """ map_func = shlex_quote if quote else lambda x: x return " ".join([map_func(c) for c in command_argv if c is not None])
def _wait(cmds, **popen_kwds): p = subprocess.Popen(cmds, **popen_kwds) stdout, stderr = p.communicate() if p.returncode != 0: raise CommandLineException(argv_to_str(cmds), stdout, stderr, p.returncode) return stdout
[docs]def download_command(url, to=STDOUT_INDICATOR, quote_url=False): """Build a command line to download a URL. By default the URL will be downloaded to standard output but a specific file can be specified with the `to` argument. """ if quote_url: url = "'%s'" % url if to != STDOUT_INDICATOR: to = "'%s'" % to if which("wget"): download_cmd = ["wget", "-q"] if to == STDOUT_INDICATOR: download_cmd.extend(["-O", STDOUT_INDICATOR, url]) else: download_cmd.extend(["--recursive", "-O", to, url]) else: download_cmd = ["curl", "-L", url] if to != STDOUT_INDICATOR: download_cmd.extend(["-o", to]) return download_cmd
[docs]class CommandLineException(Exception): """An exception indicating a non-zero command-line exit."""
[docs] def __init__(self, command, stdout, stderr, returncode): """Construct a CommandLineException from command and standard I/O.""" self.command = command self.stdout = stdout self.stderr = stderr self.returncode = returncode self.message = ("Failed to execute command-line %s, stderr was:\n" "-------->>begin stderr<<--------\n" "%s\n" "-------->>end stderr<<--------\n" "-------->>begin stdout<<--------\n" "%s\n" "-------->>end stdout<<--------\n" ) % (command, stderr, stdout)
def __str__(self): """Return a verbose error message indicating the command problem.""" return self.message
__all__ = ( 'argv_to_str', 'CommandLineException', 'download_command', 'execute', 'redirect_aware_commmunicate', 'redirecting_io', 'shell', 'shell_process', 'which', )