Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
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()
return result.strip().split('\n')
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
)
LOGGER.info('Running cdk bootstrap...')
run_module_command(cmd_list=bootstrap_command,
env_vars=self.context.env_vars)
elif command == 'destroy' and 'CI' in self.context.env_vars: # noqa
cdk_opts.append('-f') # Don't prompt
cdk_command = generate_node_command(
'cdk',
cdk_opts,
self.path
)
LOGGER.info("Running cdk %s on %s (\"%s\")",
command,
os.path.basename(self.path),
format_npm_command_for_logging(cdk_command)) # noqa
run_module_command(cmd_list=cdk_command,
env_vars=self.context.env_vars)
else:
LOGGER.info(
"Skipping cdk %s of %s; no \"package.json\" "
"file was found (need a package file specifying "
"aws-cdk in devDependencies)",
command,
Args:
command (str): The Serverless command to be executed.
args_list (Optiona[List[str]]): Additional arguments to include
in the generated command.
Returns:
List[str]: The full command to be passed into a subprocess.
"""
args = [command] + self.cli_args + self.options.args
args.extend(args_list or [])
if self.context.no_color and '--no-color' not in args:
args.append('--no-color')
if command not in ['remove', 'print'] and self.context.is_noninteractive:
args.append('--conceal') # hide secrets from serverless output
cmd = generate_node_command(command='sls',
command_opts=args,
path=self.path)
self.log_npm_command(cmd)
return cmd
func_zip = os.path.basename(key) + ".zip"
if does_s3_object_exist(bucketname, hash_zip):
LOGGER.info('Found existing package "s3://%s/%s" for %s', bucketname, hash_zip, key)
download(bucketname, hash_zip, os.path.join(package_dir, func_zip))
else:
LOGGER.info('No existing package found, uploading to s3://%s/%s', bucketname,
hash_zip)
zip_name = os.path.join(package_dir, func_zip)
upload(bucketname, hash_zip, zip_name)
sls_opts[0] = 'deploy'
# --package must be provided to "deploy" as a relative path to support
# serverless@<1.70.0. the fix to support absolute path was implimented
# somewhere between 1.60.0 and 1.70.0.
sls_opts[-1] = os.path.relpath(package_dir)
sls_deploy_cmd = generate_node_command(command='sls',
command_opts=sls_opts,
path=path)
LOGGER.info("Running sls deploy on %s (\"%s\")",
os.path.basename(path),
format_npm_command_for_logging(sls_deploy_cmd))
run_module_command(cmd_list=sls_deploy_cmd,
env_vars=context.env_vars)
shutil.rmtree(package_dir)
directory=self.path,
env=self.context.env_vars
)
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':
LOGGER.info("Running cdk %s on each stack in %s",
command,
os.path.basename(self.path))
for i in get_cdk_stacks(self.path,
self.context.env_vars,
cdk_context_opts):
subprocess.call(
generate_node_command(
'cdk',
cdk_opts + [i], # 'diff '
self.path
),
env=self.context.env_vars
)
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(
def run_sls_print(sls_opts, env_vars, path):
"""Run sls print command."""
sls_info_opts = list(sls_opts)
sls_info_opts[0] = 'print'
sls_info_opts.extend(['--format', 'yaml'])
sls_info_cmd = generate_node_command(command='sls',
command_opts=sls_info_opts,
path=path)
return yaml.safe_load(subprocess.check_output(sls_info_cmd,
env=env_vars))
sls_opts (List[str]): List of options for Serverless CLI.
bucketname (str): S3 Bucket name.
context (Context): Runway context object.
path (str): Module path.
"""
package_dir = tempfile.mkdtemp()
LOGGER.debug('Package directory: %s', package_dir)
ensure_bucket_exists(bucketname, context.env_region)
sls_config = run_sls_print(sls_opts, context.env_vars, path)
hashes = get_src_hash(sls_config, path)
sls_opts[0] = 'package'
sls_opts.extend(['--package', package_dir])
sls_package_cmd = generate_node_command(command='sls',
command_opts=sls_opts,
path=path)
LOGGER.info("Running sls package on %s (\"%s\")",
os.path.basename(path),
format_npm_command_for_logging(sls_package_cmd))
run_module_command(cmd_list=sls_package_cmd,
env_vars=context.env_vars)
for key in hashes.keys():
hash_zip = hashes[key] + ".zip"
func_zip = os.path.basename(key) + ".zip"
if does_s3_object_exist(bucketname, hash_zip):
LOGGER.info('Found existing package "s3://%s/%s" for %s', bucketname, hash_zip, key)
download(bucketname, hash_zip, os.path.join(package_dir, func_zip))
'cdk',
cdk_opts + [i], # 'diff '
self.path
),
env=self.context.env_vars
)
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
)
LOGGER.info('Running cdk bootstrap...')
run_module_command(cmd_list=bootstrap_command,
env_vars=self.context.env_vars)
elif command == 'destroy' and 'CI' in self.context.env_vars: # noqa
cdk_opts.append('-f') # Don't prompt
cdk_command = generate_node_command(
'cdk',
cdk_opts,
self.path
)
LOGGER.info("Running cdk %s on %s (\"%s\")",