You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
bazarr/libs/dynaconf/loaders/yaml_loader.py

131 lines
4.2 KiB

from __future__ import annotations
import sys
from pathlib import Path
from typing import TextIO
from warnings import warn
from dynaconf import default_settings
from dynaconf.constants import YAML_EXTENSIONS
from dynaconf.loaders.base import BaseLoader
from dynaconf.utils import object_merge
from dynaconf.utils.parse_conf import try_to_encode
from dynaconf.vendor.ruamel import yaml
# Add support for Dynaconf Lazy values to YAML dumper
yaml.SafeDumper.yaml_representers[
None
] = lambda self, data: yaml.representer.SafeRepresenter.represent_str(
self, try_to_encode(data)
)
class AllLoader(BaseLoader):
"""YAML Loader to load multi doc files"""
@staticmethod
def _assign_data(data, source_file, content):
"""Helper to iterate through all docs in a file"""
content = tuple(content)
if len(content) == 1:
data[source_file] = content[0]
elif len(content) > 1:
for i, doc in enumerate(content):
data[f"{source_file}[{i}]"] = doc
def get_source_data(self, files):
data = {}
for source_file in files:
if source_file.endswith(self.extensions):
try:
with open(source_file, **self.opener_params) as open_file:
content = self.file_reader(open_file)
self.obj._loaded_files.append(source_file)
self._assign_data(data, source_file, content)
except OSError as e:
if ".local." not in source_file:
warn(
f"{self.identifier}_loader: {source_file} "
f":{str(e)}"
)
else:
# for tests it is possible to pass string
content = self.string_reader(source_file)
self._assign_data(data, source_file, content)
return data
def load(obj, env=None, silent=True, key=None, filename=None, validate=False):
"""
Reads and loads in to "obj" a single key or all keys from source file.
:param obj: the settings instance
:param env: settings current env default='development'
:param silent: if errors should raise
:param key: if defined load a single key, else load all in env
:param filename: Optional custom filename to load
:return: None
"""
# Resolve the loaders
# https://github.com/yaml/pyyaml/wiki/PyYAML-yaml.load(input)-Deprecation
# Possible values are:
# `safe_load, full_load, unsafe_load, load, safe_load_all`
yaml_reader = getattr(
yaml, obj.get("YAML_LOADER_FOR_DYNACONF"), yaml.safe_load
)
if yaml_reader.__name__ == "unsafe_load": # pragma: no cover
warn(
"yaml.unsafe_load is deprecated."
" Please read https://msg.pyyaml.org/load for full details."
" Try to use full_load or safe_load."
)
_loader = BaseLoader
if yaml_reader.__name__.endswith("_all"):
_loader = AllLoader
loader = _loader(
obj=obj,
env=env,
identifier="yaml",
extensions=YAML_EXTENSIONS,
file_reader=yaml_reader,
string_reader=yaml_reader,
validate=validate,
)
loader.load(
filename=filename,
key=key,
silent=silent,
)
def write(settings_path, settings_data, merge=True):
"""Write data to a settings file.
:param settings_path: the filepath
:param settings_data: a dictionary with data
:param merge: boolean if existing file should be merged with new data
:param stdout: boolean if should output to stdout instead of file
"""
settings_path = Path(settings_path)
if settings_path.exists() and merge: # pragma: no cover
with open(
str(settings_path), encoding=default_settings.ENCODING_FOR_DYNACONF
) as open_file:
object_merge(yaml.safe_load(open_file), settings_data)
with open(
str(settings_path),
"w",
encoding=default_settings.ENCODING_FOR_DYNACONF,
) as open_file:
yaml.dump(
settings_data,
open_file,
Dumper=yaml.dumper.SafeDumper,
explicit_start=True,
indent=2,
default_flow_style=False,
)