Coverage for lynceus/core/config/lynceus_config.py: 99%
243 statements
« prev ^ index » next coverage.py v7.10.0, created at 2025-07-29 08:46 +0000
1import configparser
2import json
3from collections.abc import (Iterable,
4 MutableMapping)
5from configparser import ConfigParser
6from copy import deepcopy
7from json import (JSONDecodeError,
8 JSONEncoder)
9from logging import Logger
10from pathlib import Path
11from typing import Any, Callable
13from lynceus.core.config import CONFIG_JSON_DUMP_KEY_END_KEYWORD
14from lynceus.lynceus_exceptions import LynceusConfigError
15from lynceus.utils import lookup_root_path
18# pylint: disable=too-many-public-methods
19class LynceusConfig(MutableMapping):
20 """A dictionary that applies an arbitrary key-altering
21 function before accessing the keys.
23 IMPORTANT: when a key does not exist, it is automatically created to ease definition like ['X']['Y']['Z'] = 'XXX'
24 BUT: it means the 'in' operator should NOT be used, because it will always return True, because the checked key will be created if not existing,
25 for such situation, you should use the lynceus.core.config.lynceus_config.LynceusConfig.has_section() method.
26 """
28 UNDEFINED_VALUE: str = '$*_UNDEFINED_*$'
30 def __init__(self, *args, **kwargs):
31 """
32 Initialize a new LynceusConfig instance.
34 Parameters
35 ----------
36 *args
37 Positional arguments to initialize the config dictionary.
38 **kwargs
39 Keyword arguments to initialize the config dictionary.
40 """
41 self.__store = {}
42 self.update(dict(*args, **kwargs)) # use the free update to set keys
44 def __getitem__(self, key):
45 """
46 Get an item from the config, automatically creating sub-dictionaries if needed.
48 Parameters
49 ----------
50 key
51 The key to retrieve.
53 Returns
54 -------
55 dict
56 The value associated with the key.
57 """
58 # Creates automatically sub dictionary if needed.
59 if LynceusConfig._keytransform(key) not in self.__store:
60 self.__store[LynceusConfig._keytransform(key)] = {}
62 return self.__store[LynceusConfig._keytransform(key)]
64 def __setitem__(self, key, value):
65 """
66 Set an item in the config.
68 Parameters
69 ----------
70 key
71 The key to set.
72 value
73 The value to associate with the key.
74 """
75 self.__store[LynceusConfig._keytransform(key)] = LynceusConfig._valuetransform(value)
77 def __delitem__(self, key):
78 """
79 Delete an item from the config.
81 Parameters
82 ----------
83 key
84 The key to delete.
85 """
86 del self.__store[LynceusConfig._keytransform(key)]
88 def __iter__(self):
89 """
90 Iterate over the config keys.
92 Returns
93 -------
94 iterator
95 Iterator over the config keys.
96 """
97 return iter(self.__store)
99 def __len__(self):
100 """
101 Get the number of items in the config.
103 Returns
104 -------
105 int
106 Number of items in the config.
107 """
108 return len(self.__store)
110 @staticmethod
111 def _keytransform(key):
112 """
113 Transform a key before using it in the store.
115 Parameters
116 ----------
117 key
118 The key to transform.
120 Returns
121 -------
122 str
123 The transformed key.
124 """
125 return key
127 @staticmethod
128 def _valuetransform(value):
129 """
130 Transform a value before storing it.
132 Parameters
133 ----------
134 value
135 The value to transform.
137 Returns
138 -------
139 Any
140 The transformed value.
141 """
142 if isinstance(value, LynceusConfig):
143 # pylint: disable=protected-access
144 return value.__store
146 return value
148 def copy(self):
149 """
150 Create a deep copy of this LynceusConfig instance.
152 Returns
153 -------
154 LynceusConfig
155 A new LynceusConfig instance with copied data.
156 """
157 lynceus_config_copy: LynceusConfig = LynceusConfig()
158 # pylint: disable=protected-access,unused-private-member
159 lynceus_config_copy.__store = deepcopy(self.__store)
160 return lynceus_config_copy
162 def as_dict(self):
163 """
164 Convert this LynceusConfig to a standard dictionary.
166 Returns
167 -------
168 dict
169 A deep copy of the internal store as a dictionary.
170 """
171 return deepcopy(self.__store)
173 @staticmethod
174 def dump_config_parser(config_parser: ConfigParser) -> str:
175 """
176 Convert a ConfigParser to a string representation.
178 Parameters
179 ----------
180 config_parser : ConfigParser
181 The ConfigParser instance to dump.
183 Returns
184 -------
185 str
186 String representation of the configuration.
187 """
188 result: str = ''
189 for config_section in config_parser.sections():
190 result += f'[{config_section}]\n'
191 for key, value in config_parser.items(config_section):
192 result += f'\t{key}={value}\n'
193 return result
195 def get_config(self, section: str, key: str, *, default: str | int | float | object | list = UNDEFINED_VALUE) -> str | int | float | object | list:
196 """
197 Get a configuration value from the specified section and key.
199 Parameters
200 ----------
201 section : str
202 The configuration section.
203 key : str
204 The configuration key.
205 default : str or int or float or object or list, optional
206 Default value to return if the config doesn't exist.
208 Returns
209 -------
210 str or int or float or object or list
211 The configuration value or default.
213 Raises
214 ------
215 LynceusConfigError
216 If the config doesn't exist and no default is provided.
217 """
218 try:
219 return self[section][key]
220 except KeyError as error:
221 # Safe-guard: raise an error if configuration not found and no default specified.
222 if default == self.UNDEFINED_VALUE:
223 raise LynceusConfigError(f'Configuration [\'{section}\'][\'{key}\'] does not exist in your configuration.') from error
224 return default
226 def check_config_exists(self, section: str, key: str, *, transform_func=None) -> bool:
227 """
228 Check if a configuration option exists and optionally transform its value.
230 This method verifies the existence of a configuration key under the specified
231 section and can apply an optional transformation function to the value if found.
232 The transformation is applied in-place, modifying the stored configuration value.
234 Args:
235 section: Configuration section name to check
236 key: Configuration key to verify within the section
237 transform_func: Optional function to transform the value if it exists.
238 Can be used for type casting or any value processing.
239 Caller must handle any exceptions during transformation.
241 Returns:
242 bool: True if the configuration option exists, False otherwise
244 Note:
245 If transform_func is provided and the configuration exists, the transformed
246 value replaces the original value in the configuration store.
247 """
248 value = self.get_config(section, key, default=None)
249 if value is None:
250 return False
252 # Checks if there is a trasnformation to apply.
253 if transform_func:
254 # Transforms (and registers) the value for further usage.
255 self[section][key] = transform_func(value)
257 # Returns the config option exists.
258 return True
260 @staticmethod
261 def to_bool(value: str | None) -> bool | None:
262 """
263 Convert a string value to boolean.
265 Parameters
266 ----------
267 value : str or None
268 The value to convert.
270 Returns
271 -------
272 bool or None
273 True for '1', 'True', or True; False otherwise; None for None input.
274 """
275 if value is None:
276 return None
277 return value in ['1', 'True', True]
279 def is_bool_config_enabled(self, section: str, key: str, *, default: bool = False) -> bool:
280 """
281 Check if a boolean configuration option is enabled.
283 Parameters
284 ----------
285 section : str
286 The configuration section.
287 key : str
288 The configuration key.
289 default : bool, optional
290 Default value if the config doesn't exist.
292 Returns
293 -------
294 bool
295 True if the configuration is enabled, False otherwise.
296 """
297 return LynceusConfig.to_bool(self.get_config(section=section, key=key, default=default))
299 def get_config_as_dict(self, section: str, key: str):
300 """
301 Get a configuration value as a dictionary.
303 Parameters
304 ----------
305 section : str
306 The configuration section.
307 key : str
308 The configuration key.
310 Returns
311 -------
312 dict
313 The configuration value as a dictionary.
314 """
315 return self.get_config(section=section, key=key)
317 def has_section(self, section: str) -> bool:
318 """
319 Check if a configuration section exists.
321 Parameters
322 ----------
323 section : str
324 The section name to check.
326 Returns
327 -------
328 bool
329 True if the section exists, False otherwise.
330 """
331 # Important: we must check the specified section on a list of keys, and not on the result of keys(), otherwise the automatic
332 # key addition will be performed in __getitem__() method, and this method will always return True.
333 return section in list(self.keys())
335 @staticmethod
336 def __cleanse_value(value: str) -> str:
337 """
338 Remove quotes from string values.
340 Parameters
341 ----------
342 value : str
343 The value to cleanse.
345 Returns
346 -------
347 str
348 The cleansed value with quotes removed.
349 """
350 return value if not isinstance(value, str) else str(value).strip("'").strip('"')
352 @staticmethod
353 def transform_value_inner_keys(value, key_transform_func):
354 """
355 Recursively transform keys of a mapping using the provided key_transform_func.
357 Parameters
358 ----------
359 value
360 The value to transform.
361 key_transform_func
362 Function to transform keys.
364 Returns
365 -------
366 dict or Any
367 The value with transformed keys.
368 """
369 if not isinstance(value, (dict, MutableMapping)) or key_transform_func is None:
370 return value
372 return {key_transform_func(k): LynceusConfig.transform_value_inner_keys(v, key_transform_func) for k, v in value.items()}
374 @staticmethod
375 def merge_iterables(dest, src):
376 """
377 Helper function to merge iterable values.
379 Parameters
380 ----------
381 dest
382 Destination iterable to merge into.
383 src
384 Source iterable to merge from.
386 Raises
387 ------
388 NotImplementedError
389 If merging is not implemented for the destination type.
390 """
391 if isinstance(dest, list):
392 dest.extend(src)
393 elif isinstance(dest, set):
394 dest.update(src)
395 else:
396 raise NotImplementedError(f'Merging not implemented for type {type(dest)}.')
398 @staticmethod
399 # pylint: disable=too-many-branches
400 def recursive_merge(dict1: dict | MutableMapping,
401 dict2: dict | MutableMapping,
402 *,
403 override_section: bool = False,
404 override_value: bool = False,
405 key_transform_func: Callable[[str], str] | None = None):
406 """
407 Recursively merge configuration data from source into destination dictionary.
409 This method performs a deep merge of two dictionaries, preserving existing values
410 when possible and handling nested structures intelligently. It supports key
411 transformation and configurable override behavior for sections and values.
413 Args:
414 dict1: Destination dictionary to merge data into (modified in-place)
415 dict2: Source dictionary containing data to merge from
416 override_section: If True, replaces existing sections completely.
417 If False, merges section contents recursively (default)
418 override_value: If True, replaces existing non-mapping values.
419 If False, raises ValueError on conflicts (default)
420 key_transform_func: Optional function to transform keys during merge.
421 Applied to both keys and nested dictionary keys
423 Raises:
424 ValueError: When value conflicts occur and override_value is False
425 NotImplementedError: For unsupported iterable merge operations
427 Note:
428 - Automatically handles iterable values (lists, sets) by extending/updating
429 - Creates new keys that don't exist in destination
430 - Applies key transformation recursively to nested structures
431 - Special handling for LynceusConfig objects to avoid auto-creation
432 """
433 for key, value in dict2.items():
434 # Transform the key if a transformation function is provided.
435 transformed_key = key_transform_func(key) if key_transform_func else key
436 transformed_value = LynceusConfig.transform_value_inner_keys(value, key_transform_func) if key_transform_func else value
438 # Checks if the corresponding key already exists in destination dict.
439 # Important: do NOT use the in operator if first argument is a LynceusConfig, to avoid automatic creation of section here.
440 if (isinstance(dict1, LynceusConfig) and not dict1.has_section(transformed_key)) or transformed_key not in dict1:
441 dict1[transformed_key] = transformed_value
442 continue
444 # Handles iterable values that are not mappings.
445 if isinstance(dict1[transformed_key], Iterable) and not isinstance(dict1[transformed_key], (dict, MutableMapping, str)):
446 # Checks if "new" value is iterable
447 if isinstance(transformed_value, Iterable) and not isinstance(transformed_value, str):
448 LynceusConfig.merge_iterables(dict1[transformed_key], transformed_value)
449 continue
451 # Handles non-mapping values.
452 if not isinstance(dict1[transformed_key], (dict, MutableMapping)):
453 if override_value or dict1[transformed_key] == transformed_value:
454 dict1[transformed_key] = transformed_value
455 continue
456 raise ValueError(f'Conflict at key "{transformed_key}": cannot merge non-mapping values'
457 f' ({override_value=}; types {type(dict1[transformed_key])} vs {type(transformed_value)}).')
459 # Handles mapping values.
460 if isinstance(transformed_value, (dict, MutableMapping)):
461 if override_section:
462 dict1[transformed_key] = transformed_value
463 else:
464 LynceusConfig.recursive_merge(dict1[transformed_key],
465 transformed_value,
466 override_section=override_section,
467 override_value=override_value,
468 key_transform_func=key_transform_func)
469 else:
470 raise ValueError(f'Conflict at key "{transformed_key}": source value is not a mapping'
471 f' ({override_value=}; types {type(dict1[transformed_key])} vs {type(transformed_value)}).')
473 def merge(self, config_map: dict[str, Any] | MutableMapping, *,
474 set_only_as_default_if_not_exist: bool = False, override_section: bool = False, override_value: bool = False, key_transform_func=None):
475 """
476 Merge a configuration dictionary into this LynceusConfig instance.
478 This method provides flexible merging strategies for configuration data,
479 supporting both additive merging and default-only merging modes. It automatically
480 processes JSON dump parameters and handles complex nested structures.
482 Args:
483 config_map: Configuration dictionary or mapping to merge into this instance
484 set_only_as_default_if_not_exist: If True, only sets values for keys that
485 don't already exist (preserves existing values).
486 If False, performs full recursive merge (default)
487 override_section: Whether to replace existing sections completely rather
488 than merging their contents (default: False)
489 override_value: Whether to replace existing non-mapping values rather
490 than raising conflicts (default: False)
491 key_transform_func: Optional function to transform keys during merge.
492 Useful for namespacing or prefixing configuration keys.
493 Cannot be used with set_only_as_default_if_not_exist=True
495 Raises:
496 NotImplementedError: If key_transform_func is used with set_only_as_default_if_not_exist=True
498 Note:
499 - Automatically processes JSON dump parameters using special key endings
500 - In default mode, uses Python 3.9+ dictionary merge operator for efficiency
501 - In recursive mode, preserves existing nested structures while adding new data
502 """
503 # Processes automatically specified config_map to manage optional json dump options.
504 config_map = LynceusConfig.load_json_params_from_lynceus_config(config_map).as_dict()
506 if set_only_as_default_if_not_exist:
507 if key_transform_func:
508 raise NotImplementedError('key_transform_func option can not be used with set_only_as_default_if_not_exist option while merging config map.')
510 # Uses the Python 3.9 ability to update Dictionary (considering existing values as the
511 # precedence ones).
512 # TODO: it may be needed to iterate on each (key, value) pairs and recursively execute the | operator on them if value is dict-like instance ...
513 self.__store = config_map | self.__store
514 return
516 # In this version, we need to merge everything, keeping existing inner (key; value) pairs.
517 LynceusConfig.recursive_merge(self, config_map, override_section=override_section, override_value=override_value, key_transform_func=key_transform_func)
519 def merge_from_lynceus_config(self, lynceus_config, *, override_section: bool = False, override_value: bool = False):
520 """
521 Merge configuration from another LynceusConfig instance.
523 Parameters
524 ----------
525 lynceus_config : LynceusConfig
526 The LynceusConfig instance to merge from.
527 override_section : bool, optional
528 Whether to override existing sections.
529 override_value : bool, optional
530 Whether to override existing values.
531 """
532 # pylint: disable=protected-access
533 self.merge(lynceus_config.__store, override_section=override_section, override_value=override_value)
535 def merge_from_config_parser(self, config_parser: ConfigParser, *, override_section: bool = False, key_transform_func=None):
536 """
537 Merge configuration data from a ConfigParser instance.
539 This method converts ConfigParser sections and key-value pairs into dictionary
540 format and merges them into this LynceusConfig instance. It automatically cleanses
541 string values by removing quotes and supports key transformation.
543 Args:
544 config_parser: ConfigParser instance containing configuration data to merge
545 override_section: If True, replaces existing sections completely.
546 If False, merges section contents with existing data (default)
547 key_transform_func: Optional function to transform keys during merge.
548 Useful for adding prefixes or namespacing configuration keys
550 Note:
551 - Automatically removes quotes from string values during conversion
552 - All merged values override existing non-mapping values (override_value=True)
553 - ConfigParser sections become top-level keys in the LynceusConfig
554 """
555 # Merges each config parser section, as a dictionary, in this instance of Lynceus config.
556 config_parser_to_dict = {config_section: {key: self.__cleanse_value(value) for key, value in config_parser.items(config_section)}
557 for config_section in config_parser.sections()}
558 self.merge(config_parser_to_dict, override_section=override_section, override_value=True, key_transform_func=key_transform_func)
560 def lookup_configuration_file_and_update_from_it(self, config_file_name: Path | str, *,
561 must_exist: bool = True,
562 root_path: Path = Path().resolve(),
563 logger: Logger | None = None,
564 **kwargs):
565 """
566 Look up a configuration file and update this config from it.
568 Parameters
569 ----------
570 config_file_name : Path or str
571 Name of the configuration file to look up.
572 must_exist : bool, optional
573 Whether the file must exist.
574 root_path : Path, optional
575 Root path to start the lookup from.
576 logger : Logger, optional
577 Optional logger for logging messages.
578 **kwargs
579 Additional arguments to pass to update_from_configuration_file.
581 Raises
582 ------
583 FileNotFoundError
584 If the file doesn't exist and must_exist is True.
585 """
586 try:
587 # Retrieves the complete path of the configuration file, if exists.
588 config_file_path = lookup_root_path(config_file_name, root_path=root_path) / Path(config_file_name)
590 # Merges it in this instance.
591 self.update_from_configuration_file(config_file_path, **kwargs)
593 if logger:
594 logger.info(f'Successfully load and merge configuration options from file "{config_file_name}".')
595 except FileNotFoundError:
596 # Raises FileNotFoundError only if configuration file should exist.
597 if must_exist:
598 raise
600 if logger:
601 logger.info(f'Configuration file "{config_file_name}" does not exist, so it will not be loaded (nor an error because {must_exist=}).')
603 def update_from_configuration_file(self, file_path: Path, *,
604 override_section: bool = False, key_transform_func=None) -> configparser.ConfigParser:
605 """
606 Load and merge configuration from a file into this LynceusConfig instance.
608 This method reads a configuration file using ConfigParser, merges its contents
609 into this LynceusConfig instance, and returns the loaded ConfigParser for
610 additional operations such as fine-tuned logger configuration.
612 Args:
613 file_path: Path to the configuration file to load and merge
614 override_section: If True, replaces existing sections completely.
615 If False, merges section contents with existing data (default)
616 key_transform_func: Optional function to transform keys during merge.
617 Useful for adding prefixes or namespacing configuration keys
619 Returns:
620 configparser.ConfigParser: The loaded ConfigParser instance for additional operations
622 Raises:
623 FileNotFoundError: If the specified configuration file does not exist
625 Note:
626 - Uses UTF-8 encoding for file reading
627 - Automatically merges all sections from the file into this instance
628 - The returned ConfigParser can be used for logger configuration setup
629 """
630 if not file_path.exists():
631 raise FileNotFoundError(f'Specified "{file_path}" configuration file path does not exist.')
633 # Reads configuration file.
634 config_parser: configparser.ConfigParser = configparser.RawConfigParser()
635 config_parser.read(str(file_path), encoding='utf8')
637 # Merges it into this instance.
638 self.merge_from_config_parser(config_parser, override_section=override_section, key_transform_func=key_transform_func)
640 return config_parser
642 @staticmethod
643 def format_config(config: MutableMapping | dict, *,
644 obfuscation_value: str | None = None,
645 sensitive_keys: list[str] | None = None,
646 obfuscation_value_from_length: int | None = None
647 ) -> dict:
648 """
649 Format a configuration dictionary with sensitive value obfuscation.
651 Recursively processes a configuration dictionary to obfuscate sensitive
652 values based on key patterns and value length. Useful for logging and
653 debugging without exposing secrets.
655 Args:
656 config: Configuration dictionary to format
657 obfuscation_value: String to replace sensitive values (default: '<secret>')
658 sensitive_keys: List of key patterns to treat as sensitive
659 obfuscation_value_from_length: Minimum length for automatic obfuscation
661 Returns:
662 dict: Formatted configuration with obfuscated sensitive values
663 """
664 obfuscation_value = obfuscation_value or '<secret>'
665 sensitive_keys = sensitive_keys or ['secret', 'password', 'pwd', 'token']
667 def obfuscator(key, value):
668 """
669 Apply obfuscation logic to key-value pairs.
671 Args:
672 key: Configuration key to check for sensitivity
673 value: Configuration value to potentially obfuscate
675 Returns:
676 The original value or obfuscation string based on sensitivity rules
677 """
678 # Safe-guard: ensures key is iterable.
679 if not isinstance(value, Iterable):
680 return value
682 # Returns obfuscation_value if key is in any specified sensitive ones.
683 if any(sensitive_key in key for sensitive_key in sensitive_keys):
684 return obfuscation_value
686 # Returns untouched value if no length limit has been specified.
687 if obfuscation_value_from_length is None:
688 return value
690 # Returns obfuscation_value if value has specified length or greater.
691 return obfuscation_value if len(str(value)) >= obfuscation_value_from_length else value
693 def process_value(key, value):
694 """
695 Recursively process configuration values for obfuscation.
697 Args:
698 key: Configuration key being processed
699 value: Configuration value to process (dict, list, or scalar)
701 Returns:
702 Processed value with appropriate obfuscation applied
703 """
704 if isinstance(value, dict):
705 return LynceusConfig.format_config(value,
706 obfuscation_value=obfuscation_value,
707 sensitive_keys=sensitive_keys,
708 obfuscation_value_from_length=obfuscation_value_from_length)
710 if isinstance(value, list):
711 return [process_value(key, item) for item in value]
713 return obfuscator(key, value)
715 return {key: process_value(key, value) for key, value in dict(config).items()}
717 @staticmethod
718 def format_dict_to_string(dict_to_convert: MutableMapping, indentation_level: int = 0) -> str:
719 """
720 Convert a dictionary to a formatted, indented string representation.
722 Recursively formats nested dictionaries and LynceusConfig objects into
723 a hierarchical string with proper indentation. Useful for configuration
724 display and logging.
726 Args:
727 dict_to_convert: Dictionary to convert to string
728 indentation_level: Current indentation level for nested formatting
730 Returns:
731 str: Formatted string representation with proper indentation
732 """
733 return '\n'.join('\t' * indentation_level + f'{key}={str(value)}'
734 if not isinstance(value, dict) and not isinstance(value, LynceusConfig)
735 else '\t' * indentation_level + f'[{key}]\n' + LynceusConfig.format_dict_to_string(value, indentation_level + 1)
736 for key, value in dict_to_convert.items())
738 def dump_for_config_file(self) -> str:
739 """
740 Dump this config to a string suitable for saving to a configuration file.
742 Returns
743 -------
744 str
745 String representation of the configuration.
746 """
747 return LynceusConfig.dump_to_config_str(self.__store)
749 @staticmethod
750 def dump_to_config_str(value) -> str:
751 """
752 Dump a value to a JSON configuration string.
754 Parameters
755 ----------
756 value
757 The value to dump.
759 Returns
760 -------
761 str
762 JSON string representation of the value.
763 """
764 return json.dumps(value, cls=LynceusConfigJSONEncoder)
766 @staticmethod
767 def load_from_config_str(lynceus_config_dump: str | dict[str, Any]):
768 """
769 Load a LynceusConfig from a JSON configuration string.
771 Parameters
772 ----------
773 lynceus_config_dump : str or dict
774 JSON string or dict to load from.
776 Returns
777 -------
778 LynceusConfig or dict
779 Loaded configuration data.
781 Raises
782 ------
783 LynceusConfigError
784 If the JSON cannot be loaded.
785 """
786 # TODO: may be better to instantiate an empty LynceusConfig, and use merge method here too.
787 try:
788 loaded_collection = json.loads(lynceus_config_dump)
789 except JSONDecodeError as exc:
790 # pylint: disable=raise-missing-from
791 raise LynceusConfigError(f'Unable to JSON load config from specified string: "{lynceus_config_dump}".', from_exception=exc)
793 if isinstance(loaded_collection, dict):
794 return LynceusConfig(loaded_collection)
796 return loaded_collection
798 @staticmethod
799 def load_json_params_from_lynceus_config(config_map: dict[str, Any] | MutableMapping):
800 """
801 Load JSON parameters from a configuration map.
803 Parameters
804 ----------
805 config_map : dict or MutableMapping
806 Configuration map to load from.
808 Returns
809 -------
810 LynceusConfig
811 Loaded LynceusConfig instance.
813 Raises
814 ------
815 LynceusConfigError
816 If the config_map is not a dict or MutableMapping.
817 """
818 # Safe-guard: ensures specified parameter is a dict.
819 if not isinstance(config_map, (dict, MutableMapping)):
820 raise LynceusConfigError(f'Specified parameter {config_map=} should be a dict ({type(config_map)} actually.')
822 # Loads the specified config (e.g. custom user Project Profile metadata, jobs overriding/configuration options):
823 # - if key ends with the special json keyword, loads it
824 # - otherwise keep the key, value metadata pair unchanged
825 loaded_lynceus_config: LynceusConfig = LynceusConfig()
826 for config_param_with_json_key in list(config_map):
827 # Checks if it is a json dump metadata.
828 config_param_with_json_key_name: str = str(config_param_with_json_key)
829 if not config_param_with_json_key_name.endswith(CONFIG_JSON_DUMP_KEY_END_KEYWORD):
830 # It is NOT a json dump.
832 # Checks if the value is a complex one.
833 if isinstance(config_map[config_param_with_json_key_name], (dict, MutableMapping)):
834 # Calls recursively this method on the complex value.
835 loaded_lynceus_config[config_param_with_json_key_name].update(LynceusConfig.load_json_params_from_lynceus_config(config_map[config_param_with_json_key_name]).as_dict())
836 else:
837 # Registers this single value.
838 LynceusConfig.recursive_merge(loaded_lynceus_config, {config_param_with_json_key_name: config_map[config_param_with_json_key_name]})
840 continue
842 # Loads the json value, and merges it recursively.
843 loaded_metadata_key = config_param_with_json_key_name.removesuffix(CONFIG_JSON_DUMP_KEY_END_KEYWORD)
844 value_dump = config_map[config_param_with_json_key]
845 LynceusConfig.recursive_merge(loaded_lynceus_config, {loaded_metadata_key: LynceusConfig.load_from_config_str(value_dump)})
847 return loaded_lynceus_config
849 def save_partial_to_config_file(self, *, section_key_map: list[tuple], file_path: Path,
850 key_transform_func=None,
851 logger: Logger | None = None, log_prefix: str | None = ''):
852 """
853 Save specific configuration options to a configuration file.
855 This method extracts specified configuration options from this LynceusConfig
856 instance and saves them to a configuration file in standard INI format.
857 It handles different value types appropriately, using JSON serialization
858 for complex types and string representation for simple types.
860 Args:
861 section_key_map: List of (section, key) tuples specifying which configuration
862 options to save. Non-existing options are silently skipped
863 file_path: Path where the configuration file should be saved
864 key_transform_func: Optional function to transform keys before saving.
865 Useful for adding prefixes or namespacing
866 logger: Optional logger for recording save operations and debugging
867 log_prefix: Optional prefix string for log messages
869 Note:
870 - Path objects are saved as strings, not JSON
871 - Non-string complex types are saved as JSON with special key suffix
872 - String values are saved directly without modification
873 - Creates ConfigParser sections automatically as needed
874 - Uses UTF-8 encoding for file writing
875 """
876 config_parser: ConfigParser = ConfigParser()
877 not_existing_value: str = 'no/th/ing'
879 # Defines the metadata lynceus file, from complete configuration file, if defined.
880 for metadata_option in section_key_map:
881 section: str = metadata_option[0]
882 key: str = metadata_option[1]
884 # Attempts to retrieve the value.
885 value = self.get_config(section, key, default=not_existing_value)
886 if value == not_existing_value:
887 continue
889 # Ensures section exists.
890 if not config_parser.has_section(section):
891 config_parser.add_section(section)
893 # Transforms key if func is specified (useful to save project template config options, while processing the project to score).
894 if key_transform_func:
895 key = key_transform_func(key)
897 # Registers it.
898 if isinstance(value, Path):
899 # - Path are saved as a string (not as a json dump, which is not loadable).
900 value = str(value)
901 elif not isinstance(value, str):
902 # - all other type, but string, are saved as Json dump.
903 key += CONFIG_JSON_DUMP_KEY_END_KEYWORD
904 value = self.dump_to_config_str(value)
905 # - registers it inside config_parser.
906 config_parser.set(section, key, value)
908 if not config_parser.sections():
909 logger.info(f'{log_prefix} no configuration option found to create Lynceus metadata file content among "{section_key_map}"')
910 return
912 if logger:
913 logger.debug(f'{log_prefix} created Lynceus metadata file content:\n{LynceusConfig.dump_config_parser(config_parser)}')
915 with open(file_path, 'w', encoding='utf8') as file:
916 config_parser.write(file)
918 if logger:
919 logger.debug(f'{log_prefix} successfully written Lynceus metadata to file "{file_path}".')
921 def is_empty(self):
922 """
923 Check if the configuration is empty.
925 Returns
926 -------
927 bool
928 True if the configuration has no items, False otherwise.
929 """
930 return len(self.__store) == 0
932 def __repr__(self):
933 """
934 Return the string representation of this LynceusConfig.
936 Returns
937 -------
938 str
939 Formatted string representation of the configuration.
940 """
941 return str(LynceusConfig.format_config(self))
943 def __str__(self):
944 """
945 Return the string representation of this LynceusConfig.
947 Returns
948 -------
949 str
950 Formatted string representation of the configuration.
951 """
952 return LynceusConfig.format_dict_to_string(LynceusConfig.format_config(self))
954 def __or__(self, other):
955 """
956 Implement the | operator for merging configurations.
958 Parameters
959 ----------
960 other : LynceusConfig or dict
961 Another LynceusConfig or dict to merge with.
963 Returns
964 -------
965 LynceusConfig
966 New LynceusConfig instance with merged data.
968 Raises
969 ------
970 ValueError
971 If other is not a LynceusConfig or dict.
972 """
973 if isinstance(other, LynceusConfig):
974 # pylint: disable=protected-access
975 return LynceusConfig(self.__store | other.__store)
977 # Implements Python 3.9+ dict | operator allowing mixing between LynceusConfig and dict.
978 if isinstance(other, dict):
979 return LynceusConfig(self.__store | other)
981 raise ValueError(f'| operator can not be used on "{type(self)}" with'
982 f' an instance of type {type(other)}.')
985class LynceusConfigJSONEncoder(JSONEncoder):
986 """
987 Custom JSON encoder for LynceusConfig objects.
988 """
990 def default(self, o):
991 """
992 Convert objects to JSON-serializable format.
994 Parameters
995 ----------
996 o
997 Object to encode.
999 Returns
1000 -------
1001 Any
1002 JSON-serializable representation of the object.
1003 """
1004 if isinstance(o, LynceusConfig):
1005 return dict(o)
1007 if isinstance(o, set):
1008 return list(o)
1010 if isinstance(o, Path):
1011 return str(o)
1013 # Let the base class default method raise the TypeError
1014 return json.JSONEncoder.default(self, o)