Skip to content

Instantly share code, notes, and snippets.

@cluelessperson
Created May 6, 2019 06:31
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 cluelessperson/f552608ce9bc7805b85eb5fad29009ce to your computer and use it in GitHub Desktop.
Save cluelessperson/f552608ce9bc7805b85eb5fad29009ce to your computer and use it in GitHub Desktop.
import logging, pathlib, json, sys, subprocess, os
from .. import version
from .logging import root_logger as logger
class Config(object):
def __init__(self):
self.config = {"log_level": "INFO"}
def __getattr__(self, attr):
try:
return self.config[attr]
except KeyError:
return None
def __setattr__(self, attr, value):
if attr in ["log", "config"]:
super(Config, self).__setattr__(attr, value)
else:
self.config[attr] = value
class Source:
def __init__(self, parent_logger=logger):
self.log = parent_logger.getChild(self.__class__.__name__)
self.config = Config()
class FileConfigSource(Source):
def __init__(self, *args, path="config.json", required=False, **kwargs):
super(FileConfigSource, self).__init__(*args, **kwargs)
self._load_config(path, required=required)
def _load_config(self, path, required=False):
path = pathlib.Path(path)
self.log.info("Looking for config file. ({}required)- {}".format("" if required else " not", path))
if path.is_file():
self.log.info("Found. Loading.")
with open(path, 'r') as f:
config = json.load(f)
self.log.debug(config)
self.config.config.update(config)
else:
log = "Not found."
if required:
self.log.fatal(log)
raise FileNotFoundError(path)
else:
self.log.info(log)
class GCloudFileConfigSource(FileConfigSource):
def __init__(self, *args, **kwargs):
project_name = self._get_cloud_project_name()
super(GCloudFileConfigSource, self).__init__(
*args,
path="{}.json".format(project_name),
required=False,
**kwargs
)
@staticmethod
def _get_cloud_project_name():
call = "gcloud config get-value project"
result = subprocess.check_output(call.split(" "), stderr=open(os.devnull, 'wb'))
project_name = result.decode('utf-8').strip()
return project_name
class EnvironmentalVariableConfigSource(Source):
def __init__(self, *args, **kwargs):
super(EnvironmentalVariableConfigSource, self).__init__(*args, **kwargs)
environmental_variable_config = self._get_environmental_key_values()
self.log.info("Checking for environmental variables. - {}_<KEY>".format(version.NAME.upper()))
if environmental_variable_config:
self.log.info(environmental_variable_config)
self.config.config.update(environmental_variable_config)
@staticmethod
def _get_environmental_key_values():
def parse_environmental_variables():
prefix = version.NAME.upper() + "_"
for key in os.environ:
if key.startswith(prefix) and len(key) > len(prefix):
value = os.environ[key]
key = key[len(prefix):].lower()
yield key, value
return dict(parse_environmental_variables())
class CliConfigSource(Source):
def __init__(self, *args, **kwargs):
super(CliConfigSource, self).__init__(*args, **kwargs)
cli_config = self._get_cli_config()
self.log.debug(cli_config)
# if config file specified, load that
if 'config' in cli_config:
self.log.info("Config file specified. - {}".format(cli_config['config']))
file_config = FileConfigSource(path=cli_config['config'], required=True, parent_logger=self.log)
self.config.config.update(file_config.config)
self.config.config.update(cli_config)
@staticmethod
def _get_cli_config():
def get_cli_key_vals():
for i, arg in enumerate(sys.argv):
if "--" in arg and len(sys.argv)-1 > i:
key = arg[2:]
value = sys.argv[i+1]
yield key, value
return dict(get_cli_key_vals())
class SourceCombiner(Source):
def __init__(self, sources, *args, **kwargs):
super(SourceCombiner, self).__init__(*args, **kwargs)
for source in sources:
config = source(parent_logger=self.log)
self.config.config.update(config.config.config)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment