1 2
import io
2 2
import warnings
3

4 2
from dynaconf.utils import build_env_list
5 2
from dynaconf.utils import ensure_a_list
6 2
from dynaconf.utils import upperfy
7

8

9 2
class BaseLoader(object):
10
    """Base loader for dynaconf source files.
11

12
    :param obj: {[LazySettings]} -- [Dynaconf settings]
13
    :param env: {[string]} -- [the current env to be loaded defaults to
14
      [development]]
15
    :param identifier: {[string]} -- [identifier ini, yaml, json, py, toml]
16
    :param extensions: {[list]} -- [List of extensions with dots ['.a', '.b']]
17
    :param file_reader: {[callable]} -- [reads file return dict]
18
    :param string_reader: {[callable]} -- [reads string return dict]
19
    """
20

21 2
    def __init__(
22
        self, obj, env, identifier, extensions, file_reader, string_reader
23
    ):
24
        """Instantiates a loader for different sources"""
25 2
        self.obj = obj
26 2
        self.env = env or obj.current_env
27 2
        self.identifier = identifier
28 2
        self.extensions = extensions
29 2
        self.file_reader = file_reader
30 2
        self.string_reader = string_reader
31

32 2
    @staticmethod
33
    def warn_not_installed(obj, identifier):  # pragma: no cover
34
        if identifier not in obj._not_installed_warnings:
35
            warnings.warn(
36
                f"{identifier} support is not installed in your environment. "
37
                f"`pip install dynaconf[{identifier}]`"
38
            )
39
        obj._not_installed_warnings.append(identifier)
40

41 2
    def load(self, filename=None, key=None, silent=True):
42
        """
43
        Reads and loads in to `self.obj` a single key or all keys from source
44

45
        :param filename: Optional filename to load
46
        :param key: if provided load a single key
47
        :param silent: if load erros should be silenced
48
        """
49 2
        filename = filename or self.obj.get(self.identifier.upper())
50 2
        if not filename:
51 2
            return
52

53 2
        if not isinstance(filename, (list, tuple)):
54 2
            split_files = ensure_a_list(filename)
55 2
            if all([f.endswith(self.extensions) for f in split_files]):  # noqa
56 2
                files = split_files  # it is a ['file.ext', ...]
57
            else:  # it is a single config as string
58 2
                files = [filename]
59
        else:  # it is already a list/tuple
60 2
            files = filename
61

62 2
        source_data = self.get_source_date(files)
63

64 2
        if self.obj.get("ENVIRONMENTS_FOR_DYNACONF") is False:
65 2
            self._envless_load(source_data, silent, key)
66
        else:
67 2
            self._load_all_envs(source_data, silent, key)
68

69 2
    def get_source_date(self, files):
70
        """Reads each file and returns source data for each file
71
        {"path/to/file.ext": {"key": "value"}}
72
        """
73 2
        data = {}
74 2
        for source_file in files:
75 2
            if source_file.endswith(self.extensions):
76 2
                try:
77 2
                    with io.open(
78
                        source_file,
79
                        encoding=self.obj.get(
80
                            "ENCODING_FOR_DYNACONF", "utf-8"
81
                        ),
82
                    ) as open_file:
83 2
                        content = self.file_reader(open_file)
84 2
                        self.obj._loaded_files.append(source_file)
85 2
                        if content:
86 2
                            data[source_file] = content
87 2
                except IOError as e:
88 2
                    if ".local." not in source_file:
89 2
                        warnings.warn(
90
                            f"{self.identifier}_loader: {source_file} "
91
                            f":{str(e)}"
92
                        )
93
            else:
94
                # for tests it is possible to pass string
95 2
                content = self.string_reader(source_file)
96 2
                if content:
97 2
                    data[source_file] = content
98 2
        return data
99

100 2
    def _envless_load(self, source_data, silent=True, key=None):
101
        """Load all the keys from each file without env separation"""
102 2
        for source_file, file_data in source_data.items():
103 2
            self._set_data_to_obj(
104
                file_data, self.identifier, source_file, key=key
105
            )
106

107 2
    def _load_all_envs(self, source_data, silent=True, key=None):
108
        """Load configs from files separating by each environment"""
109

110 2
        for source_file, file_data in source_data.items():
111

112
            # env name is checked in lower
113 2
            file_data = {k.lower(): value for k, value in file_data.items()}
114

115
            # is there a `dynaconf_merge` on top level of file?
116 2
            file_merge = file_data.get("dynaconf_merge")
117

118
            # all lower case for comparison
119 2
            base_envs = [
120
                # DYNACONF or MYPROGRAM
121
                (self.obj.get("ENVVAR_PREFIX_FOR_DYNACONF") or "").lower(),
122
                # DEFAULT
123
                self.obj.get("DEFAULT_ENV_FOR_DYNACONF").lower(),
124
                # default active env unless ENV_FOR_DYNACONF is changed
125
                "development",
126
                # backwards compatibility for global
127
                "dynaconf",
128
                # global that rules all
129
                "global",
130
            ]
131

132 2
            for env in build_env_list(self.obj, self.env):
133 2
                env = env.lower()  # lower for better comparison
134 2
                data = {}
135 2
                try:
136 2
                    data = file_data[env] or {}
137 2
                except KeyError:
138 2
                    if env not in base_envs:
139 2
                        message = (
140
                            f"{self.identifier}_loader: {env} env not"
141
                            f"defined in {source_file}"
142
                        )
143 2
                        if silent:
144 2
                            warnings.warn(message)
145
                        else:
146 2
                            raise KeyError(message)
147 2
                    continue
148

149 2
                if env != self.obj.get("DEFAULT_ENV_FOR_DYNACONF").lower():
150 2
                    identifier = f"{self.identifier}_{env}"
151
                else:
152 2
                    identifier = self.identifier
153

154 2
                self._set_data_to_obj(
155
                    data, identifier, source_file, file_merge, key, env
156
                )
157

158 2
    def _set_data_to_obj(
159
        self,
160
        data,
161
        identifier,
162
        source_file,
163
        file_merge=None,
164
        key=False,
165
        env=False,
166
    ):
167
        """Calls setttings.set to add the keys"""
168

169
        # data 1st level keys should be transformed to upper case.
170 2
        data = {upperfy(k): v for k, v in data.items()}
171 2
        if key:
172 2
            key = upperfy(key)
173

174 2
        is_secret = "secret" in source_file
175

176
        # is there a `dynaconf_merge` inside an `[env]`?
177 2
        file_merge = file_merge or data.pop("DYNACONF_MERGE", False)
178

179 2
        if not key:
180 2
            self.obj.update(
181
                data,
182
                loader_identifier=identifier,
183
                is_secret=is_secret,
184
                merge=file_merge,
185
            )
186 2
        elif key in data:
187 2
            self.obj.set(
188
                key,
189
                data.get(key),
190
                loader_identifier=identifier,
191
                is_secret=is_secret,
192
                merge=file_merge,
193
            )

Read our documentation on viewing source code .

Loading