"""CDK module."""
import logging
import os
import subprocess
import sys
from .._logging import PrefixAdaptor
from ..util import change_dir, run_commands, which
from . import (
RunwayModule,
generate_node_command,
run_module_command,
run_npm_install,
warn_on_boto_env_vars,
)
LOGGER = logging.getLogger(__name__)
[docs]def get_cdk_stacks(module_path, env_vars, context_opts):
"""Return list of CDK stacks."""
LOGGER.debug("listing stacks in the CDK app prior to diff...")
result = subprocess.check_output(
generate_node_command(
command="cdk", command_opts=["list"] + context_opts, path=module_path
),
env=env_vars,
)
if isinstance(result, bytes): # python3 returns encoded bytes
result = result.decode()
result = result.strip().split("\n")
LOGGER.debug("found stacks: %s", result)
return result
[docs]class CloudDevelopmentKit(RunwayModule):
"""CDK Runway Module."""
def __init__(self, context, path, options=None):
"""Instantiate class.
Args:
context (Context): Runway context object.
path (Union[str, Path]): Path to the module.
options (Dict[str, Dict[str, Any]]): Everything in the module
definition merged with applicable values from the deployment
definition.
"""
super(CloudDevelopmentKit, self).__init__(context, path, options)
# logger needs to be created here to use the correct logger
self.logger = PrefixAdaptor(self.name, LOGGER)
[docs] def run_cdk(self, command="deploy"): # pylint: disable=too-many-branches
"""Run CDK."""
response = {"skipped_configs": False}
cdk_opts = [command]
if self.context.no_color:
cdk_opts.append("--no-color")
if not which("npm"):
self.logger.error(
'"npm" not found in path or is not executable; '
"please ensure it is installed correctly."
)
sys.exit(1)
if "DEBUG" in self.context.env.vars:
cdk_opts.append("-v") # Increase logging if requested
warn_on_boto_env_vars(self.context.env.vars)
if self.options["environment"]:
if os.path.isfile(os.path.join(self.path, "package.json")):
with change_dir(self.path):
run_npm_install(
self.path, self.options, self.context, logger=self.logger
)
if self.options.get("options", {}).get("build_steps", []):
self.logger.info("build steps (in progress)")
run_commands(
commands=self.options.get("options", {}).get(
"build_steps", []
),
directory=self.path,
env=self.context.env.vars,
)
self.logger.info("build steps (complete)")
cdk_context_opts = []
for (key, val) in self.options["parameters"].items():
cdk_context_opts.extend(["-c", "%s=%s" % (key, val)])
cdk_opts.extend(cdk_context_opts)
if command == "diff":
self.logger.info("plan (in progress)")
for i in get_cdk_stacks(
self.path, self.context.env.vars, cdk_context_opts
):
subprocess.call(
generate_node_command(
"cdk", cdk_opts + [i], self.path # 'diff <stack>'
),
env=self.context.env.vars,
)
self.logger.info("plan (complete)")
else:
# Make sure we're targeting all stacks
if command in ["deploy", "destroy"]:
cdk_opts.append('"*"')
if command == "deploy":
if "CI" in self.context.env.vars:
cdk_opts.append("--ci")
cdk_opts.append("--require-approval=never")
bootstrap_command = generate_node_command(
"cdk",
["bootstrap"]
+ cdk_context_opts
+ (["--no-color"] if self.context.no_color else []),
self.path,
)
self.logger.info("bootstrap (in progress)")
run_module_command(
cmd_list=bootstrap_command,
env_vars=self.context.env.vars,
logger=self.logger,
)
self.logger.info("bootstrap (complete)")
elif command == "destroy" and self.context.is_noninteractive:
cdk_opts.append("-f") # Don't prompt
cdk_command = generate_node_command("cdk", cdk_opts, self.path)
self.logger.info("%s (in progress)", command)
run_module_command(
cmd_list=cdk_command,
env_vars=self.context.env.vars,
logger=self.logger,
)
self.logger.info("%s (complete)", command)
else:
self.logger.info(
'skipped; package.json with "aws-cdk" in devDependencies is '
"required for this module type"
)
else:
self.logger.info("skipped; environment required but not defined")
response["skipped_configs"] = True
return response
[docs] def plan(self):
"""Run cdk diff."""
self.run_cdk(command="diff")
[docs] def deploy(self):
"""Run cdk deploy."""
self.run_cdk(command="deploy")
[docs] def destroy(self):
"""Run cdk destroy."""
self.run_cdk(command="destroy")