vbatts/quay
Archived
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Revert inmemoryprov, skip local storage validation

Browse source
This commit is contained in:
Sam Chow 2018-06-25 15:23:30 -04:00
parent f32bbf1fdc
commit 7619ab44e5
9 changed files with 35 additions and 115 deletions
Download patch file Download diff file Expand all files Collapse all files

3
config_app/config_util/config/__init__.py
View file

@ -1,6 +1,5 @@
from config_app.config_util.config.fileprovider import FileConfigProvider
from config_app.config_util.config.testprovider import TestConfigProvider
from config_app.config_util.config.inmemoryprovider import InMemoryProvider
def get_config_provider(config_volume, yaml_filename, py_filename, testing=False):
@ -9,4 +8,4 @@ def get_config_provider(config_volume, yaml_filename, py_filename, testing=False
if testing:
return TestConfigProvider()
return InMemoryProvider()
return FileConfigProvider(config_volume, yaml_filename, py_filename)

84
config_app/config_util/config/inmemoryprovider.py
View file

@ -1,84 +0,0 @@
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