Skip to content

Instantly share code, notes, and snippets.

Last active January 7, 2019 14:20
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save rhyst/1439943cf89852088904be67a5720e08 to your computer and use it in GitHub Desktop.
Save rhyst/1439943cf89852088904be67a5720e08 to your computer and use it in GitHub Desktop.
A script for constructing docker compose files out of complicated interdependent services
import sys
import os
import re
import yaml
import pprint
import operator
from functools import reduce
from pathlib import Path
from dotenv import load_dotenv
import argparse
import shutil
ap = argparse.ArgumentParser()
ap.add_argument("command", metavar="command", default="run", help="command to use: run, enable, disable", nargs="?")
ap.add_argument("service_name", metavar="service name", help="name of service to enable or disable", nargs="?")
ap.add_argument("--env-file", default=".env", help="path to the environment variables file")
ap.add_argument("--available-dir", default="./services-available", help="path to the available folder")
ap.add_argument("--enabled-dir", default="./services-enabled", help="path to the enabled folder")
ap.add_argument("-o", "--output", default="docker-compose.yml", help="name of the output file")
ap.add_argument("-v", "--verbose", action="store_true", help="verbose output")
args = vars(ap.parse_args())
command = args['command']
service_name = args['service_name'] if 'service_name' in args else None
env_file = os.path.abspath(str(Path('.') / args['env_file']))
available_dir = os.path.abspath(str(Path('.') / args['available_dir']))
enabled_dir = os.path.abspath(str(Path('.') / args['enabled_dir']))
output_file = os.path.abspath(str(Path('.') / args['output']))
verbose = args['verbose'] if 'verbose' in args else False
if (command == 'enable' or command == 'disable') and service_name is None:
ap.error('You must specify a service name.')
pp = pprint.PrettyPrinter(indent=2)
# Load env vars from file
# Useful Functions
def getFromDict(dataDict, mapList):
return reduce(operator.getitem, mapList, dataDict)
def setInDict(dataDict, mapList, value):
origvalue = getFromDict(dataDict, mapList[:-1])[mapList[-1]]
if isinstance(origvalue,list):
getFromDict(dataDict, mapList[:-1])[mapList[-1]] = value + origvalue
getFromDict(dataDict, mapList[:-1])[mapList[-1]] = value
def merge(destination, source):
for key, value in source.items():
if isinstance(value, dict):
node = destination.setdefault(key, {})
merge(node, value)
destination[key] = value
return destination
def replace_env(match):
return os.getenv(
def enable():
service_path = os.path.join(available_dir, service_name + ".yml")
enabled_service_path = os.path.join(enabled_dir, service_name + ".yml")
if os.path.isfile(enabled_service_path):
print('Service already enabled')
if not os.path.isfile(service_path):
print("Cannot find service: {}".format(service_path))
if not os.path.isdir(enabled_dir):
print("Cannot find enabled directory: {}".format(enabled_dir))
if os.path.islink(service_path):
if os.path.isfile(enabled_service_path):
os.symlink(os.readlink(service_path), enabled_service_path)
os.symlink(service_path, enabled_service_path)
print("Enabled {}".format(service_path))
def disable():
enabled_service_path = os.path.join(enabled_dir, service_name + ".yml")
if not os.path.isfile(enabled_service_path):
print("Service is not enabled: {}".format(service_path))
if os.path.islink(enabled_service_path):
print("Disabled {}".format(service_name))
def run():
# Gather configs
output_config = {}
configs = []
config_files = [f for f in os.listdir(config_path) if os.path.isfile(os.path.join(config_path, f))]
for config_file in config_files:
with open(os.path.join(config_path, config_file), 'r') as stream:
print('Reading {}'.format(config_file))
config_text = re.sub(r'\${(.*)}', replace_env,
name = os.path.splitext(config_file)[0]
config = yaml.load(config_text)
except Exception as e:
print('Error with "{}" config file'.format(name))
if "composer_base" in config and config["composer_base"]:
output_config = config
# Merge configs
for config in configs:
output_config = merge(output_config, config)
# Resolve additions
for config in configs:
if 'composer_compositions' in config:
for composition in config['composer_compositions']:
if composition['type'] == 'add':
setInDict(output_config, composition['to'], [ composition['value'] ])
output_config.pop('composer_base', None)
output_config.pop('composer_compositions', None)
with open(output_file, 'w+') as stream:
yaml.dump(output_config, stream, default_flow_style=False)
if verbose:
print(yaml.dump(output_config, default_flow_style=False))
if command == 'run':
if command == 'enable':
if command == 'disable':
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment