Toolset to render and manage AWS CloudFormation ( )
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

100 lines
2.6 KiB

4 years ago
import os
import copy
import logging
2 years ago
import re
4 years ago
4 years ago
def dict_merge(a, b):
"""Deep merge to allow proper inheritance for config files"""
4 years ago
if not a:
return b
if not b:
return a
if not isinstance(a, dict) or not isinstance(b, dict):
4 years ago
raise TypeError
result = copy.deepcopy(a)
for k, v in b.items():
if k in result and isinstance(result[k], dict):
result[k] = dict_merge(result[k], v)
4 years ago
result[k] = copy.deepcopy(v)
return result
def ensure_dir(path):
"""Creates dir if it does not already exist."""
if not os.path.exists(path):
def setup_logging(debug):
if debug:
our_level = logging.DEBUG
formatter = logging.Formatter(
fmt="[%(asctime)s] %(name)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
4 years ago
our_level = logging.INFO
formatter = logging.Formatter(
fmt="[%(asctime)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
4 years ago
log_handler = logging.StreamHandler()
logger = logging.getLogger("cloudbender")
return logger
def search_refs(template, attributes, mode):
"""Traverses a template and searches for any remote references and
adds them to the attributes set
if isinstance(template, dict):
for k, v in template.items():
# FortyTwo Fn::GetAtt
if k == "Fn::GetAtt" and isinstance(v, list):
if v[0] == "FortyTwo":
# CloudBender::StackRef
if k == "CloudBender::StackRef":
except KeyError:
# PipedMode Refs
if mode == "Piped" and k == "Ref" and "DoT" in v:
if isinstance(v, dict) or isinstance(v, list):
search_refs(v, attributes, mode)
elif isinstance(template, list):
for k in template:
if isinstance(k, dict) or isinstance(k, list):
search_refs(k, attributes, mode)
2 years ago
def get_s3_url(url, *args):
bucket = None
path = None
m = re.match("^(s3://)?([^/]*)(/.*)?", url)
2 years ago
bucket = m[2]
if m[3]:
path = m[3].lstrip("/")
2 years ago
path = os.path.join(path, *args)
return (bucket, path)