Coverage for src/mafw/tools/toml_tools.py: 99%
213 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-26 09:13 +0000
1# Copyright 2025–2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""
5Tools for reading, writing, and validating MAFw TOML steering files.
7:Author: Bulgheroni Antonio
8:Description: Utilities to generate and load TOML steering files and related helpers.
9"""
11import datetime
12import logging
13import os
14import re
15from collections.abc import Mapping
16from pathlib import Path, PosixPath, WindowsPath
17from typing import Any, cast
19import tomlkit
20from tomlkit import TOMLDocument, boolean, comment, document, item, nl, table
21from tomlkit.exceptions import ConvertError
22from tomlkit.items import Item, String, StringType
23from tomlkit.toml_file import TOMLFile
25import mafw.mafw_errors
26from mafw.__about__ import __version__ as version
27from mafw.db.db_configurations import default_conf
28from mafw.lazy_import import LazyImportProcessor, ProcessorClassProtocol
29from mafw.processor import Processor
30from mafw.steering.builder import SteeringBuilder, ValidationLevel
32log = logging.getLogger(__name__)
34ENV_PATTERN = re.compile(
35 r"""
36 \$\{ # opening ${
37 (?P<name>[A-Za-z_][A-Za-z0-9_]*) # variable name
38 (?:
39 (?P<op>:-|:\?) # operator (:- or :?)
40 (?P<value>[^}]*) # default or error message
41 )?
42 \} # closing }
43 """,
44 re.VERBOSE,
45)
46"""Regex matching supported environment variable expansion patterns."""
48ENV_ESCAPE_SENTINEL = '__MAFW_ENV_ESCAPE__{'
49"""Sentinel used to preserve escaped variable patterns."""
51MAX_ENV_RESOLUTION_PASSES = 10
52"""Maximum number of expansion passes applied to a single string."""
55class PathItem(String):
56 """TOML item representing a Path"""
58 def unwrap(self) -> Path: # type: ignore[override] # do not know how to do it
59 return Path(super().unwrap())
62def path_encoder(obj: Any) -> Item:
63 """Encoder for PathItem."""
64 if isinstance(obj, PosixPath):
65 return PathItem.from_raw(str(obj), type_=StringType.SLB, escape=False)
66 elif isinstance(obj, WindowsPath):
67 return PathItem.from_raw(str(obj), type_=StringType.SLL, escape=False)
68 else:
69 raise ConvertError
72tomlkit.register_encoder(path_encoder)
75def generate_steering_file(
76 output_file: Path | str,
77 processors: list[ProcessorClassProtocol] | ProcessorClassProtocol,
78 database_conf: dict[str, Any] | None = None,
79 db_engine: str = 'sqlite',
80) -> None:
81 """
82 Generates a steering file.
84 :param output_file: The output filename where the steering file will be save.
85 :type output_file: Path | str
86 :param processors: The processors list for which the steering file will be generated.
87 :type processors: list[ProcessorClassProtocol] | ProcessorClassProtocol
88 :param database_conf: The database configuration dictionary
89 :type database_conf: dict, Optional
90 :param db_engine: A string representing the DB engine to be used. Possible values are: *sqlite*, *postgresql*
91 and *mysql*.
92 :type: str
93 """
94 if isinstance(output_file, str):
95 output_file = Path(output_file)
97 doc = _new_toml_doc()
98 doc = _add_db_configuration(database_conf, db_engine=db_engine, doc=doc)
99 doc = _add_processor_parameters_to_toml_doc(processors, doc)
100 doc = _add_user_interface_configuration(doc)
102 with open(output_file, 'w') as fp:
103 tomlkit.dump(doc, fp)
106def _new_toml_doc() -> TOMLDocument:
107 doc = document()
108 doc.add(comment(f'MAFw steering file generated on {datetime.datetime.now()}'))
109 doc.add(nl())
110 doc.add(
111 comment('uncomment the line below and insert the processors you want to run from the available processor list')
112 )
113 doc.add(comment('processors_to_run = []'))
114 doc.add(nl())
115 doc.add(comment('customise the name of the analysis'))
116 doc.add('analysis_name', String.from_raw('mafw analysis', StringType.SLB))
117 doc.add('analysis_description', String.from_raw('Summing up numbers', StringType.MLB))
118 doc.add('new_only', boolean('true'))
119 doc.add('mafw_version', String.from_raw(version, StringType.SLB))
120 doc.add('create_standard_tables', boolean('true'))
121 return doc
124def _add_db_configuration(
125 database_conf: dict[str, Any] | None = None, db_engine: str = 'sqlite', doc: TOMLDocument | None = None
126) -> TOMLDocument:
127 """Add the DB configuration to the TOML document
129 The expected structure of the database_conf dictionary is one of these:
131 .. code-block:: python
133 option1 = {
134 'DBConfiguration': {
135 'URL': 'sqlite:///:memory:',
136 'parameters': {
137 'sqlite': {
138 'pragmas': {
139 'journal_mode': 'wal',
140 'cache_size': -64000,
141 'foreign_keys': 1,
142 'synchronous': 0,
143 },
144 },
145 },
146 }
147 }
149 option2 = {
150 'URL': 'sqlite:///:memory:',
151 'authentication': {
152 'method': 'env',
153 'username': 'POSTGRES_USER',
154 'password': 'POSTGRES_PASS',
155 },
156 'parameters': {
157 'postgresql': {
158 'sslmode': 'require',
159 },
160 },
161 }
163 :param database_conf: A dictionary with the database configuration. See comments above. If None, then the default
164 is used.
165 :type database_conf: dict
166 :param db_engine: The database engine. It is used only in case the provided database configuration is invalid to
167 retrieve the default configuration. Defaults to sqlite.
168 :type db_engine: str, Optional
169 :param doc: The TOML document to add the DB configuration. If None, one will be created.
170 :type doc: TOMLDocument, Optional
171 :return: The modified document.
172 :rtype: TOMLDocument
173 :raises UnknownDBEngine: if the `database_conf` is invalid and the db_engine is not yet implemented.
174 """
175 if doc is None:
176 doc = _new_toml_doc()
178 if database_conf is None:
179 if db_engine in default_conf:
180 database_conf = default_conf[db_engine]
181 else:
182 log.critical('The provided db_engine (%s) is not yet implemented', db_engine)
183 raise mafw.mafw_errors.UnknownDBEngine(f'DB engine ({db_engine} not implemented')
185 is_conf_valid = True
186 if 'DBConfiguration' in database_conf:
187 # it should be option 1. let's check if there is the URL that is required.
188 if 'URL' not in database_conf['DBConfiguration']:
189 # no URL
190 is_conf_valid = False
191 else:
192 database_conf = cast(dict[str, Any], database_conf['DBConfiguration'])
193 else:
194 # option 2
195 if 'URL' not in database_conf:
196 # no URL
197 is_conf_valid = False
199 if not is_conf_valid:
200 log.error('The provided database configuration is invalid. Adding default configuration')
201 if db_engine not in default_conf:
202 log.critical('The provided db_engine (%s) is not yet implemented', db_engine)
203 raise mafw.mafw_errors.UnknownDBEngine(f'DB engine ({db_engine} not implemented')
204 database_conf = default_conf[db_engine]
206 db_table = table()
207 for key, value in database_conf.items():
208 if key == 'authentication' and isinstance(value, dict):
209 auth_table = table()
210 auth_table.comment('Select auth method; see documentation placeholder link: DOC_LINK_PLACEHOLDER')
211 for auth_key, auth_value in value.items():
212 auth_table[auth_key] = auth_value
213 db_table['authentication'] = auth_table
214 continue
215 if key == 'parameters' and isinstance(value, dict):
216 params_table = table()
217 for backend, params in value.items():
218 backend_table = table()
219 if isinstance(params, dict):
220 for param_key, param_value in params.items():
221 if param_key == 'pragmas' and isinstance(param_value, dict):
222 pragmas_table = table()
223 pragmas_table.comment('Leave these default values, unless you know what you are doing!')
224 for pragma_key, pragma_value in param_value.items():
225 pragmas_table[pragma_key] = pragma_value
226 backend_table['pragmas'] = pragmas_table
227 else:
228 backend_table[param_key] = param_value
229 params_table[backend] = backend_table
230 db_table['parameters'] = params_table
231 continue
232 db_table[key] = value
233 if key == 'URL':
234 db_table[key].comment(
235 'Change the protocol depending on the DB type. Update this file to the path of your DB.'
236 )
237 if key == 'pragmas':
238 db_table[key].comment('Leave these default values, unless you know what you are doing!')
240 doc.add('DBConfiguration', db_table)
241 doc.add(nl())
243 return doc
246def _add_processor_parameters_to_toml_doc(
247 processors: list[ProcessorClassProtocol] | ProcessorClassProtocol, doc: TOMLDocument | None = None
248) -> TOMLDocument:
249 if not isinstance(processors, list):
250 processors = [processors]
252 if not processor_validator(processors):
253 raise TypeError('Only processor instances and classes can be accepted')
255 if doc is None:
256 doc = _new_toml_doc()
258 # add an array with all available processors
259 proc_names = []
260 for processor in processors:
261 if isinstance(processor, LazyImportProcessor):
262 proc_names.append(processor.plugin_name)
263 elif isinstance(processor, Processor):
264 proc_names.append(processor.name)
265 else:
266 proc_names.append(processor.__name__)
267 doc.add('available_processors', item(proc_names))
268 doc.add(nl())
270 for p_item in processors:
271 if isinstance(p_item, LazyImportProcessor):
272 processor_cls = p_item._load()
273 section_name = processor_cls.__name__
274 docstring = processor_cls.__doc__
275 elif isinstance(p_item, Processor):
276 processor_cls = p_item.__class__
277 section_name = p_item.name
278 docstring = p_item.__doc__
279 else:
280 processor_cls = cast(type[Processor], p_item)
281 section_name = processor_cls.__name__
282 docstring = processor_cls.__doc__
284 # create a table for the current processor
285 p_table = table()
287 if docstring:
288 lines = docstring.splitlines()
289 for line in lines: 289 ↛ 295line 289 didn't jump to line 295 because the loop on line 289 didn't complete
290 line = line.strip()
291 if line:
292 p_table.comment(line)
293 break
295 for schema in processor_cls.parameter_schema():
296 p_table[schema.name] = schema.default
297 if schema.help:
298 # starting from tomlkit 0.15, item can return an union type between Item and OutOfOrderTableProxy
299 cast(Item, p_table.value.item(schema.name)).comment(schema.help)
301 doc.add(section_name, p_table)
302 doc.add(nl())
304 return doc
307def processor_validator(processors: list[ProcessorClassProtocol]) -> bool:
308 """
309 Validates that all items in the list are valid processor instances or classes.
311 :param processors: The list of items to be validated.
312 :type processors: list[ProcessorClassProtocol]
313 :return: True if all items are valid.
314 :rtype: bool
315 """
316 return all([isinstance(p, (Processor, type(Processor), LazyImportProcessor)) for p in processors])
319def dump_processor_parameters_to_toml(
320 processors: list[ProcessorClassProtocol] | ProcessorClassProtocol, output_file: Path | str
321) -> None:
322 """
323 Dumps a toml file with processor parameters.
325 This helper function can be used when the parameters of one or many processors have to be dumped to a TOML file.
326 For each Processor in the `processors` a table in the TOML file will be added with their parameters is the shape of
327 parameter name = value.
329 It must be noted that `processors` can be:
331 - a list of processor classes (list[type[Processor]])
332 - a list of processor instances (list[Processor]])
333 - one single processor class (type[Processor])
334 - one single processor instance (Processor)
336 What value of the parameters will be dumped?
337 --------------------------------------------
339 Good question, have a look at this :ref:`explanation <parameter_dump>`.
341 :param processors: One or more processors for which the parameters should be dumped.
342 :type processors: list[ProcessorClassProtocol] | ProcessorClassProtocol
343 :param output_file: The name of the output file for the dump.
344 :type output_file: Path | str
345 :raise KeyAlreadyPresent: if an attempt to add twice, the same processor is made.
346 :raise TypeError: if the list contains items different from Processor classes and instances.
347 """
349 doc = _add_processor_parameters_to_toml_doc(processors)
351 with open(output_file, 'w') as fp:
352 tomlkit.dump(doc, fp)
355def _add_user_interface_configuration(doc: TOMLDocument | None = None) -> TOMLDocument:
356 if doc is None:
357 doc = _new_toml_doc()
359 ui_table = table()
360 ui_table.comment('Specify UI options')
361 ui_table['interface'] = 'rich'
362 ui_table['interface'].comment('Default "rich", backup "console"')
363 doc.add('UserInterface', ui_table)
365 return doc
368def load_steering_file_legacy(steering_file: Path | str) -> dict[str, Any]:
369 """
370 Load a steering file without any semantic validation.
372 :param steering_file: The path to the steering file.
373 :type steering_file: Path, str
374 :return: The parsed steering dictionary.
375 :rtype: dict
376 """
377 if isinstance(steering_file, str):
378 steering_file = Path(steering_file)
380 doc = TOMLFile(steering_file).read()
381 return doc.value
384def load_steering_file(
385 steering_file: Path | str, validation_level: ValidationLevel | None = ValidationLevel.SEMANTIC
386) -> dict[str, Any]:
387 """
388 Load a steering file for the execution framework.
390 :param steering_file: The path to the steering file.
391 :type steering_file: Path, str
392 :param validation_level: Requested validation tier, or ``None`` to skip validation.
393 :return: The configuration dictionary.
394 :rtype: dict
395 :raise mafw.mafw_errors.InvalidSteeringFile: if the validation level reports at least one issue.
396 """
397 builder = SteeringBuilder.from_toml(steering_file)
398 if validation_level is not None:
399 issues = builder.validate(validation_level)
400 if issues:
401 raise issues[0]
402 return resolve_config_env(builder.to_config_dict())
405def resolve_config_env(config: dict[str, Any], env: Mapping[str, str] | None = None) -> dict[str, Any]:
406 """
407 Resolve environment variables in every string value of a configuration dictionary.
409 :param config: Configuration dictionary to resolve.
410 :type config: dict[str, Any]
411 :param env: Optional environment mapping; defaults to ``os.environ``.
412 :type env: Mapping[str, str] | None
413 :return: A new configuration dictionary with resolved values.
414 :rtype: dict[str, Any]
415 :raises ValueError: If a required variable is missing or expansion does not converge.
416 """
417 if env is None:
418 env = os.environ
419 return cast(dict[str, Any], _resolve_value(config, env))
422def _resolve_value(value: Any, env: Mapping[str, str]) -> Any:
423 if isinstance(value, str):
424 return resolve_string(value, env)
425 if isinstance(value, dict):
426 return {key: _resolve_value(item, env) for key, item in value.items()}
427 if isinstance(value, list):
428 return [_resolve_value(item, env) for item in value]
429 return value
432def resolve_string(value: str, env: Mapping[str, str]) -> str:
433 """
434 Resolve environment variables within a string value.
436 :param value: Input string to resolve.
437 :type value: str
438 :param env: Environment mapping to use for substitution.
439 :type env: Mapping[str, str]
440 :return: The resolved string.
441 :rtype: str
442 :raises ValueError: If a required variable is missing or expansion does not converge.
443 """
444 escaped = value.replace(r'\${', ENV_ESCAPE_SENTINEL)
446 for _ in range(MAX_ENV_RESOLUTION_PASSES):
447 if not ENV_PATTERN.search(escaped):
448 break
449 escaped = ENV_PATTERN.sub(lambda match: _resolve_match(match, env), escaped)
450 else:
451 raise ValueError('Environment variable expansion did not converge.')
453 return escaped.replace(ENV_ESCAPE_SENTINEL, '${')
456def _resolve_match(match: re.Match[str], env: Mapping[str, str]) -> str:
457 name = match.group('name')
458 op = match.group('op')
459 value = match.group('value') or ''
461 if name in env:
462 return env[name]
463 if op == ':-':
464 return value
465 if op == ':?':
466 raise ValueError(value or f'{name} is required')
467 raise ValueError(f"Environment variable '{name}' not set")