azdev/utilities/command.py (63 lines of code) (raw):

# ----------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. See License.txt in the project root for # license information. # ----------------------------------------------------------------------------- import os import subprocess import sys import shlex from knack.log import get_logger from knack.util import CommandResultItem logger = get_logger(__name__) class CommandError(Exception): def __init__(self, output, exit_code, command): message = "Command `{}` failed with exit code {}:\n{}".format(command, exit_code, output) self.exit_code = exit_code self.output = output self.command = command super().__init__(message) def call(command, **kwargs): """ Run an arbitrary command but don't buffer the output. :param command: The entire command line to run. :param kwargs: Any kwargs supported by subprocess.Popen :returns: (int) process exit code. """ from azdev.utilities import IS_WINDOWS cmd_args = command if IS_WINDOWS and command.startswith('az '): cmd_args = "az.bat " + command[3:] if not IS_WINDOWS: cmd_args = shlex.split(command) return subprocess.run( cmd_args, check=False, # supress subprocess-run-check linter warning, no CalledProcessError **kwargs).returncode def cmd(command, message=False, show_stderr=True, raise_error=False, **kwargs): """ Run an arbitrary command. :param command: The entire command line to run. :param message: A custom message to display, or True (bool) to use a default. :param show_stderr: On error, display the contents of STDERR. :param raise_error: On error, raise CommandError. :param kwargs: Any kwargs supported by subprocess.Popen :returns: CommandResultItem object. """ from azdev.utilities import IS_WINDOWS, display # use default message if custom not provided if message is True: message = 'Running: {}\n'.format(command) if message: display(message) logger.info("Running: %s", command) cmd_args = command if IS_WINDOWS and command.startswith('az '): cmd_args = "az.bat " + command[3:] if not IS_WINDOWS: cmd_args = shlex.split(command) try: output = subprocess.run( cmd_args, check=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT if show_stderr else None, **kwargs).stdout.decode('utf-8').strip() logger.debug(output) return CommandResultItem(output, exit_code=0, error=None) except subprocess.CalledProcessError as err: if raise_error: raise CommandError(err.output.decode(), err.returncode, command) return CommandResultItem(err.output, exit_code=err.returncode, error=err) def py_cmd(command, message=False, show_stderr=True, raise_error=False, is_module=True, **kwargs): """ Run a script or command with Python. :param command: The arguments to run python with. :param message: A custom message to display, or True (bool) to use a default. :param show_stderr: On error, display the contents of STDERR. :param raise_error: On error, raise CommandError. :param is_module: Run a Python module as a script with -m. :param kwargs: Any kwargs supported by subprocess.Popen :returns: CommandResultItem object. """ from azdev.utilities import get_env_path env_path = get_env_path() python_bin = sys.executable if not env_path else os.path.join( env_path, 'Scripts' if sys.platform == 'win32' else 'bin', 'python') if is_module: command = '{} -m {}'.format(python_bin, command) else: command = '{} {}'.format(python_bin, command) return cmd(command, message, show_stderr, raise_error, **kwargs) def pip_cmd(command, message=False, show_stderr=True, raise_error=True, **kwargs): """ Run a pip command. :param command: The arguments to run pip with. :param message: A custom message to display, or True (bool) to use a default. :param show_stderr: On error, display the contents of STDERR. :param raise_error: On error, raise CommandError. As pip_cmd is usually called as a control function, instead of a test target, default to True. :param kwargs: Any kwargs supported by subprocess.Popen :returns: CommandResultItem object. """ command = 'pip {}'.format(command) return py_cmd(command, message, show_stderr, raise_error, **kwargs)