1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
from __future__ import annotations
from pathlib import Path
from warnings import warn
from dynaconf import default_settings
from dynaconf.constants import YAML_EXTENSIONS
from dynaconf.loaders.base import BaseLoader
from dynaconf.loaders.base import SourceMetadata
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,
identifier="yaml",
):
"""
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
# when load_file function is called directly it comes with module and line number
if isinstance(identifier, SourceMetadata) and identifier.loader.startswith(
"load_file"
):
identifier = identifier.loader
loader = _loader(
obj=obj,
env=env,
identifier=identifier,
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,
)
|
