in osci/config/reader.py [0:0]
def read_config_from_databricks_secrets(config: dict, dbutils=None) -> dict:
log.debug('Check read config from databricks secrets variables')
out_config = dict()
if dbutils is None:
log.error('`dbutils` is not defined')
return out_config
try:
scope = config[META_CONFIG_FIELD][DATABRICKS_SECRETS_SCOPE_FIELD]
def _load_variables_from_secrets(variable):
if isinstance(variable, dict):
return {
k: _load_variables_from_secrets(v) for k, v in variable.items()
}
if isinstance(variable, list):
return [_load_variables_from_secrets(v) for v in variable]
return dbutils.secrets.get(scope=scope, key=str(variable))
out_config = {k: _load_variables_from_secrets(v) for k, v in config.items() if k != META_CONFIG_FIELD}
except KeyError as ex:
log.error(f'Databricks scope field (`{DATABRICKS_SECRETS_SCOPE_FIELD}`) not found '
f'in {config[META_CONFIG_FIELD]}: {ex}')
return out_config