edq.core.config

  1import argparse
  2import os
  3import typing
  4
  5import platformdirs
  6
  7import edq.util.dirent
  8import edq.util.json
  9
 10CONFIG_SOURCE_GLOBAL: str = "<global config file>"
 11CONFIG_SOURCE_LOCAL: str = "<local config file>"
 12CONFIG_SOURCE_CLI_FILE: str = "<cli config file>"
 13CONFIG_SOURCE_CLI: str = "<cli argument>"
 14
 15CONFIG_PATHS_KEY: str = 'config_paths'
 16CONFIGS_KEY: str = 'configs'
 17GLOBAL_CONFIG_KEY: str = 'global_config_path'
 18IGNORE_CONFIGS_KEY: str = 'ignore_configs'
 19DEFAULT_CONFIG_FILENAME: str = "edq-config.json"
 20
 21class ConfigSource:
 22    """ A class for storing config source information. """
 23
 24    def __init__(self, label: str, path: typing.Union[str, None] = None) -> None:
 25        self.label = label
 26        """ The label identifying the config (see CONFIG_SOURCE_* constants). """
 27
 28        self.path = path
 29        """ The path of where the config was sourced from. """
 30
 31    def __eq__(self, other: object) -> bool:
 32        if (not isinstance(other, ConfigSource)):
 33            return False
 34
 35        return ((self.label == other.label) and (self.path == other.path))
 36
 37    def __str__(self) -> str:
 38        return f"({self.label}, {self.path})"
 39
 40def get_global_config_path(config_filename: str) -> str:
 41    """ Get the path for the global config file. """
 42
 43    return platformdirs.user_config_dir(config_filename)
 44
 45def get_tiered_config(
 46        config_filename: str = DEFAULT_CONFIG_FILENAME,
 47        legacy_config_filename: typing.Union[str, None] = None,
 48        cli_arguments: typing.Union[dict, argparse.Namespace, None] = None,
 49        local_config_root_cutoff: typing.Union[str, None] = None,
 50    ) -> typing.Tuple[typing.Dict[str, str], typing.Dict[str, ConfigSource]]:
 51    """
 52    Load all configuration options from files and command-line arguments.
 53    Returns a configuration dictionary with the values based on tiering rules and a source dictionary mapping each key to its origin.
 54    """
 55
 56    if (cli_arguments is None):
 57        cli_arguments = {}
 58
 59    config: typing.Dict[str, str] = {}
 60    sources: typing.Dict[str, ConfigSource] = {}
 61
 62    # Ensure CLI arguments are always a dict, even if provided as argparse.Namespace.
 63    if (isinstance(cli_arguments, argparse.Namespace)):
 64        cli_arguments = vars(cli_arguments)
 65
 66    global_config_path = cli_arguments.get(GLOBAL_CONFIG_KEY, get_global_config_path(config_filename))
 67
 68    # Check the global user config file.
 69    if (os.path.isfile(global_config_path)):
 70        _load_config_file(global_config_path, config, sources, CONFIG_SOURCE_GLOBAL)
 71
 72    # Check the local user config file.
 73    local_config_path = _get_local_config_path(
 74        config_filename = config_filename,
 75        legacy_config_filename = legacy_config_filename,
 76        local_config_root_cutoff = local_config_root_cutoff,
 77    )
 78
 79    if (local_config_path is not None):
 80        _load_config_file(local_config_path, config, sources, CONFIG_SOURCE_LOCAL)
 81
 82    # Check the config file specified on the command-line.
 83    config_paths = cli_arguments.get(CONFIG_PATHS_KEY, [])
 84    for path in config_paths:
 85        _load_config_file(path, config, sources, CONFIG_SOURCE_CLI_FILE)
 86
 87    # Check the command-line config options.
 88    cli_configs = cli_arguments.get(CONFIGS_KEY, [])
 89    for cli_config in cli_configs:
 90        if ("=" not in cli_config):
 91            raise ValueError(
 92                f"Invalid configuration option '{cli_config}'."
 93                + " Configuration options must be provided in the format `<key>=<value>` when passed via the CLI."
 94            )
 95
 96        (key, value) = cli_config.split("=", maxsplit = 1)
 97
 98        key = key.strip()
 99        if (key == ""):
100            raise ValueError(f"Found an empty configuration option key associated with the value '{value}'.")
101
102        config[key] = value
103        sources[key] = ConfigSource(label = CONFIG_SOURCE_CLI)
104
105    # Finally, ignore any configs that is specified from CLI command.
106    cli_ignore_configs = cli_arguments.get(IGNORE_CONFIGS_KEY, [])
107    for ignore_config in cli_ignore_configs:
108        config.pop(ignore_config, None)
109        sources.pop(ignore_config, None)
110
111    return config, sources
112
113def _load_config_file(
114        config_path: str,
115        config: typing.Dict[str, str],
116        sources: typing.Dict[str, ConfigSource],
117        source_label: str,
118    ) -> None:
119    """ Loads config variables and the source from the given config JSON file. """
120
121    config_path = os.path.abspath(config_path)
122    for (key, value) in edq.util.json.load_path(config_path).items():
123        key = key.strip()
124        if (key == ""):
125            raise ValueError(f"Found an empty configuration option key associated with the value '{value}'.")
126
127        config[key] = value
128        sources[key] = ConfigSource(label = source_label, path = config_path)
129
130def _get_local_config_path(
131        config_filename: str,
132        legacy_config_filename: typing.Union[str, None] = None,
133        local_config_root_cutoff: typing.Union[str, None] = None,
134    ) -> typing.Union[str, None]:
135    """
136    Search for a config file in hierarchical order.
137    Begins with the provided config file name,
138    optionally checks the legacy config file name if specified,
139    then continues up the directory tree looking for the provided config file name.
140    Returns the path to the first config file found.
141
142    If no config file is found, returns None.
143
144    The cutoff parameter limits the search depth, preventing detection of config file in higher-level directories during testing.
145    """
146
147    # Provided config file is in current directory.
148    if (os.path.isfile(config_filename)):
149        return os.path.abspath(config_filename)
150
151    # Provided legacy config file is in current directory.
152    if (legacy_config_filename is not None):
153        if (os.path.isfile(legacy_config_filename)):
154            return os.path.abspath(legacy_config_filename)
155
156    # Provided config file is found in an ancestor directory up to the root or cutoff limit.
157    parent_dir = os.path.dirname(os.getcwd())
158    return _get_ancestor_config_file_path(
159        parent_dir,
160        config_filename = config_filename,
161        local_config_root_cutoff = local_config_root_cutoff,
162    )
163
164def _get_ancestor_config_file_path(
165        current_directory: str,
166        config_filename: str,
167        local_config_root_cutoff: typing.Union[str, None] = None,
168    ) -> typing.Union[str, None]:
169    """
170    Search through the parent directories (until root or a given cutoff directory(inclusive)) for a config file.
171    Stops at the first occurrence of the specified config file along the path to root.
172    Returns the path if a config file is found.
173    Otherwise, returns None.
174    """
175
176    if (local_config_root_cutoff is not None):
177        local_config_root_cutoff = os.path.abspath(local_config_root_cutoff)
178
179    current_directory = os.path.abspath(current_directory)
180    for _ in range(edq.util.dirent.DEPTH_LIMIT):
181        config_file_path = os.path.join(current_directory, config_filename)
182        if (os.path.isfile(config_file_path)):
183            return config_file_path
184
185        # Check if current directory is root.
186        parent_dir = os.path.dirname(current_directory)
187        if (parent_dir == current_directory):
188            break
189
190        if (local_config_root_cutoff == current_directory):
191            break
192
193        current_directory = parent_dir
194
195    return None
196
197def set_cli_args(parser: argparse.ArgumentParser, extra_state: typing.Dict[str, typing.Any],
198        config_filename: str = DEFAULT_CONFIG_FILENAME,
199        **kwargs: typing.Any,
200    ) -> None:
201    """
202    Set common CLI arguments for configuration.
203    """
204
205    group = parser.add_argument_group('config options')
206
207    group.add_argument('--config', dest = CONFIGS_KEY,
208        action = 'append', type = str, default = [],
209        help = ('Set a configuration option from the command-line.'
210            + ' Specify options as <key>=<value> pairs.'
211            + ' This flag can be specified multiple times.'
212            + ' The options are applied in the order provided and later options override earlier ones.'
213            + ' Will override options form all config files.')
214    )
215
216    group.add_argument('--config-file', dest = CONFIG_PATHS_KEY,
217        action = 'append', type = str, default = [],
218        help = ('Load config options from a JSON file.'
219            + ' This flag can be specified multiple times.'
220            + ' Files are applied in the order provided and later files override earlier ones.'
221            + ' Will override options form both global and local config files.')
222    )
223
224    group.add_argument('--config-global', dest = GLOBAL_CONFIG_KEY,
225        action = 'store', type = str, default = get_global_config_path(config_filename),
226        help = 'Set the default global config file path (default: %(default)s).',
227    )
228
229    group.add_argument('--ignore-config-option', dest = IGNORE_CONFIGS_KEY,
230        action = 'append', type = str, default = [],
231        help = ('Ignore any config option with the specified key.'
232            + ' The system-provided default value will be used for that option if one exists.'
233            + ' This flag can be specified multiple times.'
234            + ' Ignored options are processed last.')
235    )
236
237def load_config_into_args(
238        parser: argparse.ArgumentParser,
239        args: argparse.Namespace,
240        extra_state: typing.Dict[str, typing.Any],
241        config_filename: str = DEFAULT_CONFIG_FILENAME,
242        cli_arg_config_map: typing.Union[typing.Dict[str, str], None] = None,
243        **kwargs: typing.Any,
244    ) -> None:
245    """
246    Take in args from a parser that was passed to set_cli_args(),
247    and get the tired configuration with the appropriate parameters, and attache it to args.
248
249    Arguments that appear on the CLI as flags (e.g. `--foo bar`) can be copied over to the config options via `cli_arg_config_map`.
250    The keys of `cli_arg_config_map` represent attributes in the CLI arguments (`args`),
251    while the values represent the desired config name this argument should be set as.
252    For example, a `cli_arg_config_map` of `{'foo': 'baz'}` will make the CLI argument `--foo bar`
253    be equivalent to `--config baz=bar`.
254    """
255
256    if (cli_arg_config_map is None):
257        cli_arg_config_map = {}
258
259    for (cli_key, config_key) in cli_arg_config_map.items():
260        value = getattr(args, cli_key, None)
261        if (value is not None):
262            getattr(args, CONFIGS_KEY).append(f"{config_key}={value}")
263
264    (config_dict, sources_dict) = get_tiered_config(
265        cli_arguments = args,
266        config_filename = config_filename,
267    )
268
269    setattr(args, "_config", config_dict)
270    setattr(args, "_config_sources", sources_dict)