# -*- coding: utf-8 -*-
"""Top-level package for Yapconf."""
import json
import re
import sys
import six
from box import Box
if six.PY3:
from collections.abc import MutableMapping
unicode = str
else:
from io import open
from collections import MutableMapping
# Setup feature flags for use throughout the package.
yaml_support = True
etcd_support = True
kubernetes_support = True
redis_support = True
try:
# We set safe_load, to be load because otherwise ruamel.yaml
# will throw warnings. Since we don't want that to happen, and
# we want our code to be the same whether or not PyYaml or
# ruamel.yaml is installed.
import ruamel.yaml as yaml
yaml.load = yaml.safe_load
except ImportError:
try:
import yaml
except ImportError:
yaml = None
yaml_support = False
try:
from etcd import client as etcd_client
except ImportError:
etcd_client = None
etcd_support = False
try:
from kubernetes import client as kubernetes_client
except ImportError:
kubernetes_client = None
kubernetes_support = False
from yapconf.exceptions import YapconfError # noqa: E402
from yapconf.spec import YapconfSpec # noqa: E402
__author__ = """Logan Asher Jones"""
__email__ = 'loganasherjones@gmail.com'
__version__ = '0.3.7'
FILE_TYPES = {'json', }
SUPPORTED_SOURCES = {
'dict',
'environment',
'json',
'cli',
}
ALL_SUPPORTED_SOURCES = {
'dict',
'environment',
'etcd',
'json',
'kubernetes',
'yaml',
'cli',
}
if yaml_support:
FILE_TYPES.add('yaml')
SUPPORTED_SOURCES.add('yaml')
if etcd_support:
SUPPORTED_SOURCES.add('etcd')
if kubernetes_support:
SUPPORTED_SOURCES.add('kubernetes')
__all__ = ['YapconfSpec', 'dump_data']
def change_case(s, separator='-'):
"""Changes the case to snake/kebab case depending on the separator.
As regexes can be confusing, I'll just go through this line by line as an
example with the following string: ' Foo2Boo_barBaz bat'
1. Remove whitespaces from beginning/end. => 'Foo2Boo_barBaz bat-rat'
2. Replace remaining spaces with underscores => 'Foo2Boo_barBaz_bat-rat'
3. Add underscores before capital letters => 'Foo2_Boo_bar_Baz_bat-rat'
4. Replace capital with lowercase => 'foo2_boo_bar_baz_bat-rat'
5. Underscores & hyphens become the separator => 'foo2-boo-bar-baz-bat-rat'
Args:
s (str): The original string.
separator: The separator you want to use (default '-' for kebab case).
Returns:
A snake_case or kebab-case (depending on separator)
"""
s = s.strip()
no_spaces = re.sub(' ', '_', s)
add_underscores = re.sub('(.)([A-Z][a-z]+)', r'\1_\2', no_spaces)
lowercase = re.sub('([a-z0-9])([A-Z])', r'\1_\2', add_underscores).lower()
return re.sub('[-_]', separator, lowercase)
[docs]def dump_data(data,
filename=None,
file_type='json',
klazz=YapconfError,
open_kwargs=None,
dump_kwargs=None):
"""Dump data given to file or stdout in file_type.
Args:
data (dict): The dictionary to dump.
filename (str, optional): Defaults to None. The filename to write
the data to. If none is provided, it will be written to STDOUT.
file_type (str, optional): Defaults to 'json'. Can be any of
yapconf.FILE_TYPES
klazz (optional): Defaults to YapconfError a special error to throw
when something goes wrong.
open_kwargs (dict, optional): Keyword arguments to open.
dump_kwargs (dict, optional): Keyword arguments to dump.
"""
_check_file_type(file_type, klazz)
open_kwargs = open_kwargs or {'encoding': 'utf-8'}
dump_kwargs = dump_kwargs or {}
if filename:
with open(filename, 'w', **open_kwargs) as conf_file:
_dump(data, conf_file, file_type, **dump_kwargs)
else:
_dump(data, sys.stdout, file_type, **dump_kwargs)
def _dump(data, stream, file_type, **kwargs):
if not kwargs and file_type == 'json':
kwargs = {
'sort_keys': True,
'indent': 4,
'ensure_ascii': False,
}
elif not kwargs and file_type == 'yaml':
kwargs = {
'default_flow_style': False,
'encoding': 'utf-8'
}
if isinstance(data, Box):
data = data.to_dict()
if str(file_type).lower() == 'json':
dumped = json.dumps(data, **kwargs)
if isinstance(dumped, unicode):
stream.write(dumped)
else:
stream.write(six.u(dumped))
elif str(file_type).lower() == 'yaml':
yaml.safe_dump(data, stream, **kwargs)
else:
raise NotImplementedError('Someone forgot to implement dump for file '
'type: %s' % file_type)
def load_file(filename,
file_type='json',
klazz=YapconfError,
open_kwargs=None,
load_kwargs=None):
"""Load a file with the given file type.
Args:
filename (str): The filename to load.
file_type (str, optional): Defaults to 'json'. The file type for the
given filename. Supported types are ``yapconf.FILE_TYPES```
klazz (optional): The custom exception to raise if something goes
wrong.
open_kwargs (dict, optional): Keyword arguments for the open call.
load_kwargs (dict, optional): Keyword arguments for the load call.
Raises:
klazz: If no klazz was passed in, this will be the ``YapconfError``
Returns:
dict: The dictionary from the file.
"""
_check_file_type(file_type, klazz)
open_kwargs = open_kwargs or {'encoding': 'utf-8'}
load_kwargs = load_kwargs or {}
data = None
with open(filename, **open_kwargs) as conf_file:
if str(file_type).lower() == 'json':
data = json.load(conf_file, **load_kwargs)
elif str(file_type).lower() == 'yaml':
data = yaml.safe_load(conf_file.read())
else:
raise NotImplementedError('Someone forgot to implement how to '
'load a %s file_type.' % file_type)
if not isinstance(data, dict):
raise klazz('Successfully loaded %s, but the result was '
'not a dictionary.' % filename)
return data
def _check_file_type(file_type, klazz):
if str(file_type).lower() == 'yaml' and yaml is None:
raise klazz('You wanted to use a YAML file but the yaml module was '
'not loaded. Please install the yaml dependency via `pip '
'install yapconf[yaml]`.')
if str(file_type).lower() not in FILE_TYPES:
raise klazz('Invalid file type %s. Valid file types are %s' %
(file_type, FILE_TYPES))
def flatten(dictionary, separator='.', prefix=''):
"""Flatten the dictionary keys are separated by separator
Arguments:
dictionary {dict} -- The dictionary to be flattened.
Keyword Arguments:
separator {str} -- The separator to use (default is '.'). It will
crush items with key conflicts.
prefix {str} -- Used for recursive calls.
Returns:
dict -- The flattened dictionary.
"""
new_dict = {}
for key, value in dictionary.items():
new_key = prefix + separator + key if prefix else key
if isinstance(value, MutableMapping):
new_dict.update(flatten(value, separator, new_key))
elif isinstance(value, list):
new_value = []
for item in value:
if isinstance(item, MutableMapping):
new_value.append(flatten(item, separator, new_key))
else:
new_value.append(item)
new_dict[new_key] = new_value
else:
new_dict[new_key] = value
return new_dict