84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
import logging
|
|
import yaml
|
|
import io
|
|
import os
|
|
|
|
from config_app.config_util.config.baseprovider import BaseProvider
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CONFIG_FILENAME = 'config.yaml'
|
|
|
|
class InMemoryProvider(BaseProvider):
|
|
def __init__(self):
|
|
self.files = {}
|
|
self.config = {}
|
|
self.was_loaded = False
|
|
|
|
@property
|
|
def provider_id(self):
|
|
return 'memory'
|
|
|
|
def update_app_config(self, app_config):
|
|
self.config = app_config
|
|
|
|
def get_config(self):
|
|
return self.config
|
|
|
|
def save_config(self, config_object):
|
|
self.config = config_object
|
|
self.was_loaded = True
|
|
|
|
def config_exists(self):
|
|
return self.was_loaded
|
|
|
|
def volume_exists(self):
|
|
return True
|
|
|
|
def volume_file_exists(self, filename):
|
|
return any([name.startswith(filename) for name in self.files])
|
|
|
|
def get_volume_file(self, filename, mode='r'):
|
|
return io.BytesIO(self.files[filename])
|
|
|
|
def write_volume_file(self, filename, contents):
|
|
raise Exception('Not implemented yet')
|
|
|
|
def remove_volume_file(self, filename):
|
|
raise Exception('Not implemented yet')
|
|
|
|
def list_volume_directory(self, path):
|
|
def strip_directory(string):
|
|
if '/' in string:
|
|
return string[string.rfind('/') + 1:]
|
|
return string
|
|
|
|
return [strip_directory(name) for name in self.files if name.startswith(path)]
|
|
|
|
def save_volume_file(self, filename, flask_file):
|
|
self.files[filename] = flask_file.read()
|
|
|
|
def requires_restart(self, app_config):
|
|
raise Exception('Not implemented yet')
|
|
|
|
def get_volume_path(self, directory, filename):
|
|
return os.path.join(directory, filename)
|
|
|
|
def load_from_tarball(self, tarfile):
|
|
for tarinfo in tarfile.getmembers():
|
|
if tarinfo.isfile():
|
|
self.files[tarinfo.name] = tarfile.extractfile(tarinfo.name).read()
|
|
|
|
if self.files.has_key(CONFIG_FILENAME):
|
|
self.config = yaml.load(self.files.get(CONFIG_FILENAME))
|
|
self.was_loaded = True
|
|
|
|
def load_from_tar_stream(self, tarfile):
|
|
for tarinfo in tarfile:
|
|
if tarinfo.isfile():
|
|
self.files[tarinfo.name] = tarfile.extractfile(tarinfo).read()
|
|
|
|
if self.files.has_key(CONFIG_FILENAME):
|
|
self.config = yaml.load(self.files.get(CONFIG_FILENAME))
|
|
self.was_loaded = True
|