139 lines
4 KiB
Python
139 lines
4 KiB
Python
import os
|
|
import yaml
|
|
import logging
|
|
import json
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def _import_yaml(config_obj, config_file):
|
|
with open(config_file) as f:
|
|
c = yaml.safe_load(f)
|
|
if not c:
|
|
logger.debug('Empty YAML config file')
|
|
return
|
|
|
|
if isinstance(c, str):
|
|
raise Exception('Invalid YAML config file: ' + str(c))
|
|
|
|
for key in c.iterkeys():
|
|
if key.isupper():
|
|
config_obj[key] = c[key]
|
|
|
|
return config_obj
|
|
|
|
|
|
def _export_yaml(config_obj, config_file):
|
|
with open(config_file, 'w') as f:
|
|
f.write(yaml.safe_dump(config_obj, encoding='utf-8', allow_unicode=True))
|
|
|
|
|
|
class BaseProvider(object):
|
|
""" A configuration provider helps to load, save, and handle config override in the application.
|
|
"""
|
|
|
|
def update_app_config(self, app_config):
|
|
""" Updates the given application config object with the loaded override config. """
|
|
raise NotImplementedError
|
|
|
|
def get_yaml(self):
|
|
""" Returns the contents of the YAML config override file, or None if none. """
|
|
raise NotImplementedError
|
|
|
|
def save_yaml(self, config_object):
|
|
""" Updates the contents of the YAML config override file to those given. """
|
|
raise NotImplementedError
|
|
|
|
def yaml_exists(self):
|
|
""" Returns true if a YAML config override file exists in the config volume. """
|
|
raise NotImplementedError
|
|
|
|
def volume_exists(self):
|
|
""" Returns whether the config override volume exists. """
|
|
raise NotImplementedError
|
|
|
|
def volume_file_exists(self, filename):
|
|
""" Returns whether the file with the given name exists under the config override volume. """
|
|
raise NotImplementedError
|
|
|
|
def save_volume_file(self, filename, flask_file):
|
|
""" Saves the given flask file to the config override volume, with the given
|
|
filename.
|
|
"""
|
|
raise NotImplementedError
|
|
|
|
|
|
class FileConfigProvider(BaseProvider):
|
|
""" Implementation of the config provider that reads the data from the file system. """
|
|
def __init__(self, config_volume, yaml_filename, py_filename):
|
|
self.config_volume = config_volume
|
|
self.yaml_filename = yaml_filename
|
|
self.py_filename = py_filename
|
|
|
|
self.yaml_path = os.path.join(config_volume, yaml_filename)
|
|
self.py_path = os.path.join(config_volume, py_filename)
|
|
|
|
def update_app_config(self, app_config):
|
|
if os.path.exists(self.py_path):
|
|
logger.debug('Applying config file: %s', self.py_path)
|
|
app_config.from_pyfile(self.py_path)
|
|
|
|
if os.path.exists(self.yaml_path):
|
|
logger.debug('Applying config file: %s', self.yaml_path)
|
|
_import_yaml(app_config, self.yaml_path)
|
|
|
|
def get_yaml(self):
|
|
if not os.path.exists(self.yaml_path):
|
|
return None
|
|
|
|
config_obj = {}
|
|
_import_yaml(config_obj, self.yaml_path)
|
|
return config_obj
|
|
|
|
def save_yaml(self, config_obj):
|
|
_export_yaml(config_obj, self.yaml_path)
|
|
|
|
def yaml_exists(self):
|
|
return self.volume_file_exists(self.yaml_filename)
|
|
|
|
def volume_exists(self):
|
|
return os.path.exists(self.config_volume)
|
|
|
|
def volume_file_exists(self, filename):
|
|
return os.path.exists(os.path.join(self.config_volume, filename))
|
|
|
|
def save_volume_file(self, filename, flask_file):
|
|
flask_file.save(os.path.join(self.config_volume, filename))
|
|
|
|
|
|
class TestConfigProvider(BaseProvider):
|
|
""" Implementation of the config provider for testing. Everything is kept in-memory instead on
|
|
the real file system. """
|
|
def __init__(self):
|
|
self.files = {}
|
|
|
|
def update_app_config(self, app_config):
|
|
pass
|
|
|
|
def get_yaml(self):
|
|
if not 'config.yaml' in self.files:
|
|
return None
|
|
|
|
return json.loads(self.files.get('config.yaml', '{}'))
|
|
|
|
def save_yaml(self, config_obj):
|
|
self.files['config.yaml'] = json.dumps(config_obj)
|
|
|
|
def yaml_exists(self):
|
|
return 'config.yaml' in self.files
|
|
|
|
def volume_exists(self):
|
|
return True
|
|
|
|
def volume_file_exists(self, filename):
|
|
return filename in self.files
|
|
|
|
def save_volume_file(self, filename, flask_file):
|
|
self.files[filename] = ''
|
|
|
|
def reset_for_test(self):
|
|
self.files = {}
|