Source code for runway.cfngin.hooks.command

"""Command hook."""
import logging
import os
import subprocess
from typing import Any, Dict, List, Optional, Union

from typing_extensions import TypedDict

from ...utils import BaseModel
from ..exceptions import ImproperlyConfigured

LOGGER = logging.getLogger(__name__)

[docs]class RunCommandHookArgs(BaseModel): """Hook arguments for ``run_command``.""" capture: bool = False """If enabled, capture the command's stdout and stderr, and return them in the hook result.""" command: Union[str, List[str]] """Command(s) to run.""" env: Optional[Dict[str, str]] = None """Dictionary of environment variable overrides for the command context. Will be merged with the current environment. """ ignore_status: bool = False """Don't fail the hook if the command returns a non-zero status.""" interactive: bool = False """If enabled, allow the command to interact with stdin. Otherwise, stdin will be set to the null device. """ quiet: bool = False """Redirect the command's stdout and stderr to the null device, silencing all output. Should not be enabled if ``capture`` is also enabled. """ stdin: Optional[str] = None """String to send to the stdin of the command. Implicitly disables ``interactive``."""
[docs]class RunCommandResponseTypeDef(TypedDict, total=False): """Response from run_command.""" returncode: int stderr: str stdout: str
[docs]def run_command(*__args: Any, **kwargs: Any) -> RunCommandResponseTypeDef: """Run a custom command as a hook. Arguments not parsed by the data model will be forwarded to the ``subprocess.Popen`` function. Interesting ones include: ``cwd`` and ``shell``. Examples: .. code-block:: yaml pre_deploy: command_copy_environment: path: runway.cfngin.hooks.command.run_command required: true enabled: true data_key: copy_env args: command: ['cp', 'environment.template', 'environment'] command_git_rev_parse: path: runway.cfngin.hooks.command.run_command required: true enabled: true data_key: get_git_commit args: command: ['git', 'rev-parse', 'HEAD'] cwd: ./my-git-repo capture: true command_npm_install: path: runway.cfngin.hooks.command.run_command args: command: '`cd $PROJECT_DIR/project; npm install`' env: PROJECT_DIR: ./my-project shell: true """ args = RunCommandHookArgs.parse_obj(kwargs) # remove parsed args from kwargs for field in RunCommandHookArgs.__fields__: kwargs.pop(field, None) # remove unneeded args from kwargs kwargs.pop("context", None) kwargs.pop("provider", None) if args.quiet and args.capture: raise ImproperlyConfigured( __name__ + ".run_command", ValueError("Cannot enable `quiet` and `capture` options simultaneously"), ) with open(os.devnull, "wb") as devnull: if args.quiet: out_err_type = devnull elif args.capture: out_err_type = subprocess.PIPE else: out_err_type = None if args.interactive: in_type = None elif args.stdin: in_type = subprocess.PIPE else: in_type = devnull if args.env: full_env = os.environ.copy() full_env.update(args.env) args.env = full_env"running command: %s", args.command) with subprocess.Popen( args.command, stdin=in_type, stdout=out_err_type, stderr=out_err_type, env=args.env, **kwargs, ) as proc: try: out, err = proc.communicate(args.stdin) status = proc.wait() if status == 0 or args.ignore_status: return {"returncode": proc.returncode, "stdout": out, "stderr": err} # Don't print the command line again if we already did earlier if LOGGER.isEnabledFor(logging.INFO): # cov: ignore LOGGER.warning("command failed with returncode %d", status) else: LOGGER.warning( "command failed with returncode %d: %s", status, args.command ) return {} except Exception: # pylint: disable=broad-except # cov: ignore return {}