Source code for dynaconf.loaders.vault_loader
# docker run -e 'VAULT_DEV_ROOT_TOKEN_ID=myroot' -p 8200:8200 vault
# pip install hvac
from dynaconf.utils.parse_conf import parse_conf_data
try:
from hvac import Client
except ImportError:
raise ImportError(
"vault package is not installed in your environment. "
"`pip install dynaconf[vault]` or disable the vault loader with "
"export VAULT_ENABLED_FOR_DYNACONF=false"
)
IDENTIFIER = 'vault'
def _get_env_list(obj, env):
"""Creates the list of environments to read
:param obj: the settings instance
:param env: settings env default='DYNACONF'
:return: a list of working environments
"""
# add the [default] env
env_list = [obj.get('DEFAULT_ENV_FOR_DYNACONF')]
# compatibility with older versions that still uses [dynaconf] as
# [default] env
global_env = obj.get('GLOBAL_ENV_FOR_DYNACONF')
if global_env not in env_list:
env_list.append(global_env)
# add the current env
if obj.current_env and obj.current_env not in env_list:
env_list.append(obj.current_env)
# add a manually set env
if env and env not in env_list:
env_list.append(env)
# add the [global] env
env_list.append('GLOBAL')
return [env.lower() for env in env_list]
[docs]def get_client(obj):
client = Client(
**{k: v for k, v in obj.VAULT_FOR_DYNACONF.items() if v is not None}
)
assert client.is_authenticated(), (
"Vault authentication error is VAULT_TOKEN_FOR_DYNACONF defined?"
)
return client
[docs]def load(obj, env=None, silent=None, key=None):
"""Reads and loads in to "settings" a single key or all keys from vault
:param obj: the settings instance
:param env: settings env default='DYNACONF'
:param silent: if errors should raise
:param key: if defined load a single key, else load all in env
:return: None
"""
client = get_client(obj)
env_list = _get_env_list(obj, env)
for env in env_list:
path = '/'.join([obj.VAULT_PATH_FOR_DYNACONF, env]).replace('//', '/')
data = client.read(path)
if data:
# There seems to be a data dict within a data dict,
# extract the inner data
data = data.get('data', {}).get('data', {})
try:
if data and key:
value = parse_conf_data(data.get(key), tomlfy=True)
if value:
obj.logger.debug(
"vault_loader: loading by key: %s:%s (%s:%s)",
key,
'****',
IDENTIFIER,
path
)
obj.set(key, value)
elif data:
obj.logger.debug(
"vault_loader: loading: %s (%s:%s)",
list(data.keys()),
IDENTIFIER,
path
)
obj.update(data, loader_identifier=IDENTIFIER, tomlfy=True)
except Exception as e:
if silent:
if hasattr(obj, 'logger'):
obj.logger.error(str(e))
return False
raise
[docs]def write(obj, data=None, **kwargs):
"""Write a value in to loader source
:param obj: settings object
:param data: vars to be stored
:param kwargs: vars to be stored
:return:
"""
if obj.VAULT_ENABLED_FOR_DYNACONF is False:
raise RuntimeError(
'Vault is not configured \n'
'export VAULT_ENABLED_FOR_DYNACONF=true\n'
'and configure the VAULT_FOR_DYNACONF_* variables'
)
data = data or {}
data.update(kwargs)
if not data:
raise AttributeError('Data must be provided')
client = get_client(obj)
path = '/'.join([obj.VAULT_PATH_FOR_DYNACONF,
obj.current_env.lower()]).replace('//', '/')
client.write(path, data=data)
load(obj)