Toolset to render and manage AWS CloudFormation ( https://pypi.org/project/cloudbender )
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.
117 lines
3.0 KiB
117 lines
3.0 KiB
import os
|
|
import copy
|
|
import logging
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
|
|
|
|
def get_docker_version():
|
|
p = shutil.which("podman")
|
|
if not p:
|
|
p = shutil.which("docker")
|
|
if not p:
|
|
return None
|
|
|
|
proc = subprocess.Popen(
|
|
[p, "--version"], stdout=subprocess.PIPE, stderr=subprocess.DEVNULL
|
|
)
|
|
if not proc.returncode:
|
|
return proc.communicate()[0].decode().strip()
|
|
else:
|
|
return None
|
|
|
|
|
|
def dict_merge(a, b):
|
|
"""Deep merge to allow proper inheritance for config files"""
|
|
if not a:
|
|
return b
|
|
|
|
if not b:
|
|
return a
|
|
|
|
if not isinstance(a, dict) or not isinstance(b, dict):
|
|
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)
|
|
else:
|
|
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):
|
|
os.makedirs(path)
|
|
|
|
|
|
def setup_logging(debug):
|
|
if debug:
|
|
our_level = logging.DEBUG
|
|
logging.getLogger("botocore").setLevel(logging.INFO)
|
|
|
|
formatter = logging.Formatter(
|
|
fmt="[%(asctime)s] %(name)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
else:
|
|
our_level = logging.INFO
|
|
logging.getLogger("botocore").setLevel(logging.CRITICAL)
|
|
|
|
formatter = logging.Formatter(
|
|
fmt="[%(asctime)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
|
|
)
|
|
|
|
log_handler = logging.StreamHandler()
|
|
log_handler.setFormatter(formatter)
|
|
logger = logging.getLogger("cloudbender")
|
|
logger.addHandler(log_handler)
|
|
logger.setLevel(our_level)
|
|
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":
|
|
attributes.append(v[1])
|
|
|
|
# CloudBender::StackRef
|
|
if k == "CloudBender::StackRef":
|
|
try:
|
|
attributes.append(v["StackTags"]["Artifact"])
|
|
except KeyError:
|
|
pass
|
|
|
|
# PipedMode Refs
|
|
if mode == "Piped" and k == "Ref" and "DoT" in v:
|
|
attributes.append(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)
|
|
|
|
|
|
def get_s3_url(url, *args):
|
|
bucket = None
|
|
path = None
|
|
|
|
m = re.match("^(s3://)?([^/]*)(/.*)?", url)
|
|
bucket = m[2]
|
|
if m[3]:
|
|
path = m[3].lstrip("/")
|
|
|
|
path = os.path.join(path, *args)
|
|
|
|
return (bucket, path)
|
|
|