"""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
LOGGER.info("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 {}