edq.core.config

  1import argparse
  2import os
  3import typing
  4
  5import platformdirs
  6
  7import edq.util.dirent
  8import edq.util.json
  9
 10CONFIG_SOURCE_CLI: str = "<cli argument>"
 11CONFIG_SOURCE_CLI_FILE: str = "<cli config file>"
 12CONFIG_SOURCE_GLOBAL: str = "<global config file>"
 13CONFIG_SOURCE_LOCAL: str = "<local config file>"
 14
 15CONFIG_PATHS_KEY: str = 'config_paths'
 16GLOBAL_CONFIG_KEY: str = 'global_config_path'
 17CONFIG_OPTIONS_KEY: str = 'configs'
 18IGNORE_CONFIG_OPTIONS_KEY: str = 'ignore_configs'
 19
 20DEFAULT_CONFIG_FILENAME: str = "edq-config.json"
 21
 22_config_filename: str = DEFAULT_CONFIG_FILENAME  # pylint: disable=invalid-name
 23_legacy_config_filename: typing.Union[str, None] = None  # pylint: disable=invalid-name
 24
 25class ConfigSource:
 26    """ A class for storing config source information. """
 27
 28    def __init__(self, label: str, path: typing.Union[str, None] = None) -> None:
 29        self.label = label
 30        """ The label identifying the config (see CONFIG_SOURCE_* constants). """
 31
 32        self.path = path
 33        """ The path of where the config was sourced from. """
 34
 35    def __eq__(self, other: object) -> bool:
 36        if (not isinstance(other, ConfigSource)):
 37            return False
 38
 39        return ((self.label == other.label) and (self.path == other.path))
 40
 41    def __str__(self) -> str:
 42        return f"({self.label}, {self.path})"
 43
 44class TieredConfigInfo(edq.util.json.DictConverter):
 45    """ A class for storing config information read from a hierarchy of files and sources. """
 46
 47    def __init__(self,
 48            config_filename: str,
 49            local_config_path: str,
 50            global_config_path: str,
 51            config: typing.Dict[str, typing.Any],
 52            sources: typing.Dict[str, ConfigSource],
 53            ) -> None:
 54        self.config_filename: str = config_filename
 55        """ Config filename searched for. """
 56
 57        self.local_config_path: str  = local_config_path
 58        """
 59        Path searched for local config.
 60        The file might not exist.
 61        """
 62
 63        self.global_config_path: str = global_config_path
 64        """
 65        Path searched for global config.
 66        The file might not exist.
 67        """
 68
 69        self.config: typing.Dict[str, typing.Any] = config
 70        """ Key-value configurations. """
 71
 72        self.sources: typing.Dict[str, ConfigSource] = sources
 73        """ Where configs came from. """
 74
 75def set_config_filename(filename: str) -> None:
 76    """ Sets the config filename. """
 77
 78    global _config_filename  # pylint: disable=global-statement
 79    _config_filename = filename
 80
 81def set_legacy_config_filename(legacy_filename: str) -> None:
 82    """ Sets the legacy config filename. """
 83
 84    global _legacy_config_filename  # pylint: disable=global-statement
 85    _legacy_config_filename = legacy_filename
 86
 87def get_config_filename() -> str:
 88    """ Gets the config filename. """
 89
 90    return _config_filename
 91
 92def get_legacy_config_filename() -> typing.Union[str, None]:
 93    """ Gets the config legacy filename. """
 94
 95    return _legacy_config_filename
 96
 97def get_global_config_path() -> str:
 98    """ Get the path for the global config file. """
 99
100    return platformdirs.user_config_dir(get_config_filename())
101
102def resolve_config_location(
103        config_info: TieredConfigInfo,
104        is_local: bool,
105        is_global: bool,
106        config_file_path: typing.Union[str, None],
107        ) -> str:
108    """
109    Resolve the config location from the given scope information.
110    Defaults to local config location if unspecified.
111    Raises an exception if an unknown config scope is given.
112    """
113
114    # Default to the local configuration if no configuration type is specified.
115    if ((not is_local) and (not is_global) and (config_file_path is None)):
116        is_local = True
117
118    if (config_file_path is not None):
119        return config_file_path
120
121    if (is_global):
122        return config_info.global_config_path
123
124    if (is_local):
125        local_config_path = config_info.local_config_path
126
127        # Fall back to the default config file name if no local config exists.
128        if (local_config_path is None):
129            local_config_path = config_info.config_filename
130
131        return local_config_path
132
133    raise ValueError("Unknown config location (e.g., not local or global).")
134
135def update_options_in_config_file(path: str, config_to_write: typing.Dict[str, str]) -> None:
136    """
137    Write configs to the specified path.
138    Create the path if it does not exist.
139    Existing keys in the file will be overwritten with the new values.
140    """
141
142    config = {}
143    if (edq.util.dirent.exists(path)):
144        config = edq.util.json.load_path(path)
145
146    config.update(config_to_write)
147
148    edq.util.dirent.mkdir(os.path.dirname(path))
149    edq.util.json.dump_path(config, path, indent = 4)
150
151def remove_options_in_config_file(path: str, config_to_remove: typing.List[str]) -> None:
152    """
153    Remove configs from the specified path.
154    Raises an exception if the given path doesn't exist.
155    """
156
157    config = edq.util.json.load_path(path)
158    for config_option in config_to_remove:
159        config.pop(config_option, None)
160
161    edq.util.json.dump_path(config, path, indent = 4)
162
163def get_tiered_config(
164        cli_arguments: typing.Union[dict, argparse.Namespace, None] = None,
165        local_config_root_cutoff: typing.Union[str, None] = None,
166        ) -> TieredConfigInfo:
167    """
168    Load all configuration options from files and command-line arguments.
169    """
170
171    if (cli_arguments is None):
172        cli_arguments = {}
173
174    config: typing.Dict[str, typing.Any] = {}
175    sources: typing.Dict[str, ConfigSource] = {}
176
177    # Ensure CLI arguments are always a dict,
178    # even if provided as argparse.Namespace.
179    if (isinstance(cli_arguments, argparse.Namespace)):
180        cli_arguments = vars(cli_arguments)
181
182    # Load the global user config file.
183    global_config_path = cli_arguments.get(GLOBAL_CONFIG_KEY, get_global_config_path())
184    _load_config_file(global_config_path, config, sources, CONFIG_SOURCE_GLOBAL)
185
186    # Get and load local user config path.
187    local_config_path = _get_local_config_path(
188        local_config_root_cutoff = local_config_root_cutoff,
189    )
190
191    if (local_config_path is None):
192        local_config_path = os.path.abspath(get_config_filename())
193
194    _load_config_file(local_config_path, config, sources, CONFIG_SOURCE_LOCAL)
195
196    # Check the config file specified on the command-line.
197    config_paths = cli_arguments.get(CONFIG_PATHS_KEY, [])
198    for path in config_paths:
199        if (not os.path.exists(path)):
200            raise FileNotFoundError(f"Specified config file does not exist: '{path}'.")
201
202        _load_config_file(path, config, sources, CONFIG_SOURCE_CLI_FILE)
203
204    # Check the command-line config options.
205    cli_configs = cli_arguments.get(CONFIG_OPTIONS_KEY, [])
206    for cli_config_option in cli_configs:
207        (key, value) = parse_string_config_option(cli_config_option)
208
209        config[key] = value
210        sources[key] = ConfigSource(label = CONFIG_SOURCE_CLI)
211
212    # Finally, ignore any configs that is specified from CLI command.
213    cli_ignore_configs = cli_arguments.get(IGNORE_CONFIG_OPTIONS_KEY, [])
214    for ignore_config in cli_ignore_configs:
215        config.pop(ignore_config, None)
216        sources.pop(ignore_config, None)
217
218    return TieredConfigInfo(get_config_filename(), local_config_path, global_config_path, config, sources)
219
220def parse_string_config_option(config_option: str) -> typing.Tuple[str, str]:
221    """
222    Parse and validate a configuration option string in the format of '<key>=<value>'.
223    Returns the resulting config option as a key-value pair.
224    """
225
226    if ("=" not in config_option):
227        raise ValueError(
228            f"Invalid configuration option string '{config_option}'."
229            + " Configuration options must be provided in the format '<key>=<value>'.")
230
231    (key, value) = config_option.split('=', maxsplit = 1)
232    key = _validate_config_key(key, value)
233
234    return key, value
235
236def _validate_config_key(config_key: str, config_value: str) -> str:
237    """ Validate a configuration key and return its clean version. """
238
239    key = config_key.strip()
240    if (key == ''):
241        raise ValueError(f"Found an empty configuration option key associated with the value '{config_value}'.")
242
243    return key
244
245def _load_config_file(
246        config_path: str,
247        config: typing.Dict[str, typing.Any],
248        sources: typing.Dict[str, ConfigSource],
249        source_label: str,
250        ) -> None:
251    """
252    Loads config variables and the source from the given config JSON file.
253    If the given config JSON file doesn't exit loads nothing.
254    """
255
256    if (not edq.util.dirent.exists(config_path)):
257        return
258
259    if (os.path.isdir(config_path)):
260        raise IsADirectoryError(f"Failed to read config file, expected a file but got a directory at '{config_path}'.")
261
262    config_path = os.path.abspath(config_path)
263    for (key, value) in edq.util.json.load_path(config_path).items():
264        key = _validate_config_key(key, value)
265
266        config[key] = value
267        sources[key] = ConfigSource(label = source_label, path = config_path)
268
269def _get_local_config_path(
270        local_config_root_cutoff: typing.Union[str, None] = None,
271        ) -> typing.Union[str, None]:
272    """
273    Search for a config file in hierarchical order.
274    Begins with the provided config file name,
275    then legacy config file name (if set),
276    then continues up the directory tree looking for the provided config file name.
277    Returns the absolute path to the first config file found.
278
279    Returns None if no config file is found.
280
281    The cutoff parameter limits the search depth,
282    preventing detection of config file in higher-level directories during testing.
283    """
284
285    config_filename = get_config_filename()
286    legacy_config_filename = get_legacy_config_filename()
287
288    # Provided config file is in current directory.
289    if (os.path.isfile(config_filename)):
290        return os.path.abspath(config_filename)
291
292    # Provided legacy config file is in current directory.
293    if (legacy_config_filename is not None):
294        if (os.path.isfile(legacy_config_filename)):
295            return os.path.abspath(legacy_config_filename)
296
297    # Provided config file is found in an ancestor directory up to the root or cutoff limit.
298    parent_dir = os.path.dirname(os.getcwd())
299    return _get_ancestor_config_file_path(
300        parent_dir,
301        local_config_root_cutoff = local_config_root_cutoff,
302    )
303
304def _get_ancestor_config_file_path(
305        current_directory: str,
306        local_config_root_cutoff: typing.Union[str, None] = None,
307        ) -> typing.Union[str, None]:
308    """
309    Search through the parent directories (until root or a given cutoff directory(inclusive)) for a config file.
310    Stops at the first occurrence of the specified config file along the path to root.
311    Returns the path if a config file is found.
312    Otherwise, returns None.
313    """
314
315    if (local_config_root_cutoff is not None):
316        local_config_root_cutoff = os.path.abspath(local_config_root_cutoff)
317
318    current_directory = os.path.abspath(current_directory)
319    for _ in range(edq.util.dirent.DEPTH_LIMIT):
320        config_file_path = os.path.join(current_directory, get_config_filename())
321        if (os.path.isfile(config_file_path)):
322            return config_file_path
323
324        # Check if current directory is root.
325        parent_dir = os.path.dirname(current_directory)
326        if (parent_dir == current_directory):
327            break
328
329        if (local_config_root_cutoff == current_directory):
330            break
331
332        current_directory = parent_dir
333
334    return None
335
336def set_cli_args(
337        parser: argparse.ArgumentParser,
338        extra_state: typing.Dict[str, typing.Any],
339        **kwargs: typing.Any,
340        ) -> None:
341    """
342    Set common CLI arguments for configuration.
343    """
344
345    group = parser.add_argument_group('config options')
346
347    group.add_argument('--config', dest = CONFIG_OPTIONS_KEY, metavar = "<KEY>=<VALUE>",
348        action = 'append', type = str, default = [],
349        help = ('Set a configuration option from the command-line.'
350            + ' Specify options as <key>=<value> pairs.'
351            + ' This flag can be specified multiple times.'
352            + ' The options are applied in the order provided and later options override earlier ones.'
353            + ' Will override options form all config files.')
354    )
355
356    group.add_argument('--config-file', dest = CONFIG_PATHS_KEY,
357        action = 'append', type = str, default = [],
358        help = ('Load config options from a JSON file.'
359            + ' This flag can be specified multiple times.'
360            + ' Files are applied in the order provided and later files override earlier ones.'
361            + ' Will override options form both global and local config files.')
362    )
363
364    group.add_argument('--config-global', dest = GLOBAL_CONFIG_KEY,
365        action = 'store', type = str, default = get_global_config_path(),
366        help = 'Set the default global config file path (default: %(default)s).',
367    )
368
369    group.add_argument('--ignore-config-option', dest = IGNORE_CONFIG_OPTIONS_KEY,
370        action = 'append', type = str, default = [],
371        help = ('Ignore any config option with the specified key.'
372            + ' The system-provided default value will be used for that option if one exists.'
373            + ' This flag can be specified multiple times.'
374            + ' Ignored options are processed last.')
375    )
376
377def add_config_location_argument_group(parser: argparse.ArgumentParser) -> None:
378    """ Add the configuration location argument group to the parser. """
379
380    group = parser.add_argument_group("config location options").add_mutually_exclusive_group()
381
382    group.add_argument('--local',
383        action = 'store_true', dest = 'scope_local',
384        help = ("Target config option(s) in a local config file.")
385    )
386
387    group.add_argument('--global',
388        action = 'store_true', dest = 'scope_global',
389        help =  ("Target config option(s) in the global config file."),
390    )
391
392    group.add_argument('--file', metavar = "<FILE>",
393        action = 'store', type = str, default = None, dest = 'scope_file',
394        help = ("Target config option(s) in a specified config file.")
395    )
396
397def load_config_into_args(
398        parser: argparse.ArgumentParser,
399        args: argparse.Namespace,
400        extra_state: typing.Dict[str, typing.Any],
401        cli_arg_config_map: typing.Union[typing.Dict[str, str], None] = None,
402        **kwargs: typing.Any,
403        ) -> None:
404    """
405    Take in args from a parser that was passed to set_cli_args(),
406    and get the tired configuration with the appropriate parameters, and attache it to args.
407
408    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`.
409    The keys of `cli_arg_config_map` represent attributes in the CLI arguments (`args`),
410    while the values represent the desired config name this argument should be set as.
411    For example, a `cli_arg_config_map` of `{'foo': 'baz'}` will make the CLI argument `--foo bar`
412    be equivalent to `--config baz=bar`.
413    """
414
415    if (cli_arg_config_map is None):
416        cli_arg_config_map = {}
417
418    for (cli_key, config_key) in cli_arg_config_map.items():
419        value = getattr(args, cli_key, None)
420        if (value is not None):
421            getattr(args, CONFIG_OPTIONS_KEY).append(f"{config_key}={value}")
422
423    config_info = get_tiered_config(cli_arguments = args)
424    setattr(args, "_config_info", config_info)