edq.config.load

  1import argparse
  2import os
  3import typing
  4
  5import edq.config.app
  6import edq.config.constants
  7import edq.config.settings
  8import edq.config.source
  9import edq.config.util
 10import edq.util.dirent
 11import edq.util.json
 12import edq.util.serial
 13
 14class ConfigLoadResult(edq.util.serial.DictConverter):
 15    """ An instance of a config value being loaded from some config source. """
 16
 17    def __init__(self,
 18            value: edq.util.serial.PODType,
 19            spec: edq.config.source.ConfigSourceSpec,
 20            path: typing.Union[str, None] = None,
 21            ) -> None:
 22        self.value: edq.util.serial.PODType = value
 23        """ The raw config value loaded. """
 24
 25        self.spec: edq.config.source.ConfigSourceSpec = spec
 26        """ The config source spec that caused this item to be loaded. """
 27
 28        self.path: typing.Union[str, None] = path
 29        """ If this result was loaded from a file, this is the path to that file. """
 30
 31    def __repr__(self) -> str:
 32        if (self.path is not None):
 33            return f"<{self.spec.label} ({self.path})>"
 34
 35        return f"<{self.spec.label}>"
 36
 37class TieredConfigInfo(edq.util.serial.DictConverter):
 38    """ A class for storing config information read from a hierarchy of files and sources. """
 39
 40    def __init__(self,
 41            raw_config: typing.Dict[str, edq.util.serial.PODType],
 42            sources: typing.Dict[str, typing.List[ConfigLoadResult]],
 43            application_config: typing.Union[edq.config.app.BaseApplicationConfig, None] = None,
 44            ) -> None:
 45        self.raw_config: typing.Dict[str, edq.util.serial.PODType] = raw_config
 46        """ The key-value config pairs after all overrides are processed. """
 47
 48        self.sources: typing.Dict[str, typing.List[ConfigLoadResult]] = sources
 49        """
 50        A representation of every source/value loaded for each config key.
 51        As config values are loaded for a key, they are pushed (not appended) onto its dictionary value.
 52        The *first* value in the list represents the final loaded value (and should match the corresponding entry in self.raw_config).
 53        """
 54
 55        if (application_config is None):
 56            application_config = edq.config.app.BaseApplicationConfig()
 57
 58        self.application_config: edq.config.app.BaseApplicationConfig = application_config
 59        """ The config typed for the specific application. """
 60
 61def get_tiered_config(
 62        cli_arguments: typing.Union[dict, argparse.Namespace, None] = None,
 63        cli_default_values: typing.Union[typing.Dict[str, typing.Any], None] = None,
 64        load_order: typing.Union[typing.List[edq.config.source.ConfigSourceSpec], None] = None,
 65        serialization_context: typing.Union[edq.util.serial.SerializationContext, None] = None,
 66        ) -> TieredConfigInfo:
 67    """
 68    Load all configuration options from files and command-line arguments.
 69    """
 70
 71    if (cli_arguments is None):
 72        cli_arguments = {}
 73
 74    if (cli_default_values is None):
 75        cli_default_values = {}
 76
 77    if (load_order is None):
 78        load_order = edq.config.settings.get_load_order()
 79
 80    raw_config: typing.Dict[str, edq.util.serial.PODType] = {}
 81    sources: typing.Dict[str, typing.List[ConfigLoadResult]] = {}
 82
 83    # Ensure CLI arguments are always a dict,
 84    # even if provided as an argparse.Namespace.
 85    if (isinstance(cli_arguments, argparse.Namespace)):
 86        cli_arguments = vars(cli_arguments)
 87
 88    # Load from each specified source.
 89    for spec in load_order:
 90        if (isinstance(spec, edq.config.source.CLIExplicitSpec)):
 91            for cli_config_option in cli_arguments.get(edq.config.constants.CONFIG_OPTIONS_KEY, []):
 92                (key, value) = edq.config.util.parse_string_config_option(cli_config_option)
 93
 94                raw_config[key] = value
 95
 96                if (key not in sources):
 97                    sources[key] = []
 98
 99                sources[key].insert(0, ConfigLoadResult(value, spec))
100        elif (isinstance(spec, edq.config.source.CLIImplicitSpec)):
101            for (key, value) in cli_arguments.items():
102                if (key in edq.config.constants.IGNORE_CLI_KEYS):
103                    continue
104
105                # Don't override existing values with default values.
106                if ((key in sources) and (value == cli_default_values.get(key, None))):
107                    continue
108
109                raw_config[key] = value
110
111                if (key not in sources):
112                    sources[key] = []
113
114                sources[key].insert(0, ConfigLoadResult(value, spec))
115        elif (isinstance(spec, edq.config.source.CLIFileSpec)):
116            config_paths = cli_arguments.get(edq.config.constants.CONFIG_PATHS_KEY, [])
117            for path in config_paths:
118                if (not os.path.exists(path)):
119                    raise FileNotFoundError(f"Specified config file does not exist: '{path}'.")
120
121                _load_config_file(path, raw_config, sources, spec)
122        elif (isinstance(spec, edq.config.source.ENVSpec)):
123            _load_env_variables(raw_config, sources, spec)
124        elif (isinstance(spec, edq.config.source.GlobalSpec)):
125            path = spec.resolve_path(override_path = cli_arguments.get(edq.config.constants.GLOBAL_CONFIG_KEY, None))
126            _load_config_file(path, raw_config, sources, spec)
127        elif (isinstance(spec, edq.config.source.AbstractPathSpec)):
128            _load_config_file(spec.resolve_path(), raw_config, sources, spec)
129        else:
130            raise ValueError(f"Unknown config source spec: '{type(spec)}'.")
131
132    if (serialization_context is None):
133        serialization_context = edq.util.serial.SerializationContext()
134    else:
135        serialization_context = serialization_context.copy()
136
137    raw_application_config = raw_config.copy()
138    encryption_key = raw_application_config.get(
139            edq.config.constants.CONFIG_ENCRYPTION_KEY,
140            edq.config.settings.get_default_encryption_key())
141    if (encryption_key is None):
142        encryption_key = edq.config.settings.get_default_encryption_key()
143
144    raw_application_config[edq.config.constants.CONFIG_ENCRYPTION_KEY] = str(encryption_key)
145    serialization_context.key = str(encryption_key)
146
147    application_config = edq.config.settings.get_application_config_class().from_dict(
148        raw_application_config,
149        context = serialization_context,
150    )
151
152    return TieredConfigInfo(
153        raw_config,
154        sources,
155        application_config = application_config,
156    )
157
158def _load_config_file(
159        config_path: str,
160        config: typing.Dict[str, typing.Any],
161        sources: typing.Dict[str, typing.List[ConfigLoadResult]],
162        spec: edq.config.source.ConfigSourceSpec,
163        ) -> None:
164    """
165    Loads config variables and the source from the given config JSON file.
166    If the given config JSON file doesn't exit loads nothing.
167    """
168
169    if (not edq.util.dirent.exists(config_path)):
170        return
171
172    if (os.path.isdir(config_path)):
173        raise IsADirectoryError(f"Failed to read config file, expected a file but got a directory at '{config_path}'.")
174
175    config_path = os.path.abspath(config_path)
176    for (key, value) in edq.util.json.load_path(config_path).items():
177        key = edq.config.util.validate_config_key(key, value)
178
179        config[key] = value
180
181        if (key not in sources):
182            sources[key] = []
183
184        sources[key].insert(0, ConfigLoadResult(value, spec, config_path))
185
186def _load_env_variables(
187        config: typing.Dict[str, typing.Any],
188        sources: typing.Dict[str, typing.List[ConfigLoadResult]],
189        spec: edq.config.source.ConfigSourceSpec,
190        ) -> None:
191    """
192    Load config from environmental variables.
193    Any variable with a matching prefix will have the prefix removed and lower-cased.
194    """
195
196    prefix = edq.config.settings.get_env_prefix()
197    for (key, value) in os.environ.items():
198        if (not key.startswith(prefix)):
199            continue
200
201        key = key.removeprefix(prefix).lower()
202
203        config[key] = value
204
205        if (key not in sources):
206            sources[key] = []
207
208        sources[key].insert(0, ConfigLoadResult(value, spec))