Coverage for src / mafw / steering / builder.py: 100%
419 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-12 09:03 +0000
1# Copyright 2026 European Union
2# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
3# SPDX-License-Identifier: EUPL-1.2
4"""Editable steering configuration builder independent from the execution engine.
6:Author: Bulgheroni Antonio
7:Description: Provides helpers to construct steering metadata without running processors.
8"""
10from __future__ import annotations
12from collections.abc import Iterable
13from enum import Enum
14from pathlib import Path
15from typing import Any, Mapping, cast
17import tomlkit
18from tomlkit.items import Array, Item, Table
19from tomlkit.toml_file import TOMLFile
21import mafw.mafw_errors
22from mafw.db.db_configurations import DEFAULT_SQLITE_PRAGMAS
23from mafw.db.db_filter import ExprParser
25from .models import (
26 Condition,
27 ConditionalFilterConfig,
28 DBConfiguration,
29 FieldFilterConfig,
30 FilterConfig,
31 GlobalSettings,
32 GroupConfig,
33 ModelFilterConfig,
34 ParameterConfig,
35 ParameterSchemaStatus,
36 ParameterSource,
37 ProcessorConfig,
38 ProcessorRef,
39 UIConfiguration,
40)
43class ValidationLevel(Enum):
44 """Validation tiers that can be requested from the steering builder."""
46 SEMANTIC = 'semantic'
47 FULL = 'full'
50class SteeringBuilder:
51 """Editable domain model for building MAFw steering files."""
53 def __init__(self) -> None:
54 self.globals = GlobalSettings()
55 self.processors: dict[str, ProcessorConfig] = {}
56 self.groups: dict[str, GroupConfig] = {}
57 self.db_config = DBConfiguration()
58 self.ui_config = UIConfiguration()
59 self._extra_globals: dict[str, Any] = {}
60 self._document: tomlkit.TOMLDocument | None = None
61 self.set_default()
63 @classmethod
64 def from_toml(cls, path: Path | str) -> 'SteeringBuilder':
65 """Create a builder from an existing steering file while keeping TOML metadata."""
67 if isinstance(path, str):
68 path = Path(path)
69 doc = TOMLFile(path).read()
70 builder = cls()
71 builder._document = doc
72 builder._load_from_document(doc)
73 return builder
75 @classmethod
76 def from_toml_text(cls, text: str) -> 'SteeringBuilder':
77 """Create a builder from TOML text while keeping TOML metadata."""
79 doc = tomlkit.parse(text)
80 builder = cls()
81 builder._document = doc
82 builder._load_from_document(doc)
83 return builder
85 def _load_from_document(self, doc: tomlkit.TOMLDocument) -> None:
86 self._extra_globals.clear()
87 # we assume that there is no DBConfiguration section.
88 self.disable_db_configuration()
90 for key, value in doc.items():
91 if key == 'processors_to_run':
92 self.globals.processors_to_run = self._ensure_str_list(value)
93 continue
94 if key in ('analysis_name', 'analysis_description', 'new_only', 'create_standard_tables'):
95 setattr(self.globals, key, self._toml_to_python(value))
96 continue
97 if key == 'DBConfiguration':
98 # we got a configuration section for the DB,
99 # we need to enable it (done inside the _parse_db_config
100 self._parse_db_config(value)
101 continue
102 if key == 'UserInterface':
103 self._parse_ui_config(value)
104 continue
105 if isinstance(value, Table):
106 parts = key.split('.')
107 if '__filter__' in parts:
108 self._parse_filter_section(parts, value)
109 continue
110 if 'processors_to_run' in value:
111 self._parse_group(key, value)
112 continue
113 self._parse_processor(key, value)
114 continue
115 self._extra_globals[key] = self._toml_to_python(value)
117 def _ensure_str_list(self, value: Any) -> list[str]:
118 python_value = self._toml_to_python(value)
119 if python_value is None:
120 return []
121 return [str(item) for item in python_value]
123 def _toml_to_python(self, value: Any) -> Any:
124 if isinstance(value, Table):
125 return {k: self._toml_to_python(v) for k, v in value.items()}
126 if isinstance(value, Array):
127 return [self._toml_to_python(item) for item in value]
128 if isinstance(value, Item):
129 return value.unwrap()
130 return value
132 def _parse_db_config(self, table: Table) -> None:
133 self.enable_db_configuration()
134 self.db_config.attributes.clear()
135 self.db_config.pragmas.clear()
136 self.db_config.authentication.clear()
137 self.db_config.parameters.clear()
138 self.db_config.url = None
140 for key, value in table.items():
141 if key == 'pragmas' and isinstance(value, Table):
142 self.db_config.pragmas = self._toml_to_python(value)
143 continue
144 if key == 'authentication' and isinstance(value, Table):
145 self.db_config.authentication = self._toml_to_python(value)
146 continue
147 if key == 'parameters' and isinstance(value, Table):
148 self.db_config.parameters = self._toml_to_python(value)
149 # we keep pragmas in a separate dictionary of the db configuration.
150 # not sure it is really needed, but handy.
151 pragmas = self.db_config.parameters.get('sqlite', {}).get('pragmas')
152 if isinstance(pragmas, dict):
153 self.db_config.pragmas = dict(pragmas)
154 continue
155 python_value = self._toml_to_python(value)
156 self.db_config.attributes[key] = python_value
157 if key == 'URL':
158 self.db_config.url = python_value
160 def _parse_ui_config(self, table: Table) -> None:
161 interface = table.get('interface')
162 if interface is not None:
163 self.ui_config.interface = self._toml_to_python(interface)
165 def _parse_group(self, name: str, table: Table) -> None:
166 processors = self._ensure_str_list(table.get('processors_to_run'))
167 description = self._toml_to_python(table.get('description'))
168 attributes: dict[str, Any] = {}
169 for attr_key, attr_value in table.items():
170 if attr_key in {'processors_to_run', 'description'}:
171 continue
172 attributes[attr_key] = self._toml_to_python(attr_value)
173 self.groups[name] = GroupConfig(
174 name=name,
175 processors=processors,
176 description=description,
177 attributes=attributes,
178 )
180 def _parse_processor(self, name: str, table: Table) -> None:
181 config = ProcessorConfig(name=name)
182 for key, value in table.items():
183 if key == '__filter__' and isinstance(value, Table):
184 logic_str = value.get('__logic__')
185 if logic_str is not None:
186 self._set_logic_expression(config, self._toml_to_python(logic_str))
187 config.filters = self._load_filters(value)
188 non_table_entries = any(not isinstance(item, Table) for item in value.values())
189 config.has_filter_root = non_table_entries
190 continue
191 if key == '__logic__':
192 self._set_logic_expression(config, self._toml_to_python(value))
193 continue
194 if key == '__new_only__':
195 config.new_only = bool(self._toml_to_python(value))
196 continue
197 if key == '__inheritance__':
198 config.inheritance = bool(self._toml_to_python(value))
199 continue
200 config.parameters[key] = ParameterConfig(
201 name=key,
202 value=self._toml_to_python(value),
203 source=ParameterSource.CONFIG,
204 status=ParameterSchemaStatus.OK,
205 )
206 self.processors[name] = config
208 def _parse_condition(self, value: Any) -> Condition:
209 if isinstance(value, dict) and 'op' in value and 'value' in value:
210 return Condition(operator=value['op'], value=value['value'], is_implicit=False)
211 if isinstance(value, list):
212 return Condition(operator='IN', value=value, is_implicit=True)
213 if isinstance(value, str):
214 return Condition(operator='GLOB', value=value, is_implicit=True)
215 return Condition(operator='==', value=value, is_implicit=True)
217 def _load_filters(self, table: Table) -> dict[str, list[FilterConfig]]:
218 result: dict[str, list[FilterConfig]] = {}
219 for model_name, model_table in table.items():
220 if model_name == '__logic__':
221 continue
222 if not isinstance(model_table, Table):
223 continue
225 # Convert TOML table to python dict first to simplify processing
226 model_data = self._toml_to_python(model_table)
227 filters: list[FilterConfig] = []
229 # 1. Model Filter (base conditions)
230 model_filter = ModelFilterConfig(name=model_name, model=model_name)
231 if '__enable__' in model_data:
232 model_filter.enabled = bool(model_data.pop('__enable__'))
234 # Extract logic for the model
235 if '__logic__' in model_data:
236 self._set_logic_expression(model_filter, model_data.pop('__logic__'))
238 # Extract conditionals
239 if '__conditional__' in model_data:
240 conditionals = model_data.pop('__conditional__')
241 if isinstance(conditionals, list):
242 for cond_data in conditionals:
243 filters.append(self._parse_conditional_config(model_name, cond_data))
245 # Process remaining fields
246 for field_name, field_value in model_data.items():
247 if isinstance(field_value, dict) and not ('op' in field_value and 'value' in field_value):
248 # It's a Field Filter (sub-table logic)
249 filters.append(self._parse_field_filter_config(model_name, field_name, field_value))
250 else:
251 # It's a simple condition for the Model Filter
252 model_filter.conditions[field_name] = self._parse_condition(field_value)
254 filters.insert(0, model_filter)
255 result[model_name] = filters
257 return result
259 def _parse_conditional_config(self, model_name: str, data: dict[str, Any]) -> ConditionalFilterConfig:
260 config = ConditionalFilterConfig(
261 name=data.get('name', ''),
262 model=model_name,
263 auto_named='name' not in data,
264 condition_field=data.get('condition_field', ''),
265 then_field=data.get('then_field', ''),
266 else_field=data.get('else_field'),
267 )
268 if '__enable__' in data:
269 config.enabled = bool(data.get('__enable__'))
271 # Only create Condition objects if sufficient data is present
272 # Or blindly create them but assume they will be serialized only if present?
273 # The test provided incomplete data, which means we should be robust.
275 if 'condition_op' in data or 'condition_value' in data:
276 config.condition = Condition(data.get('condition_op', '=='), data.get('condition_value'), is_implicit=False)
278 if 'then_op' in data or 'then_value' in data:
279 config.then_clause = Condition(data.get('then_op', '=='), data.get('then_value'), is_implicit=False)
281 if 'else_op' in data or 'else_value' in data:
282 config.else_clause = Condition(data.get('else_op', '=='), data.get('else_value'), is_implicit=False)
284 return config
286 def _parse_field_filter_config(self, model_name: str, field_name: str, data: dict[str, Any]) -> FieldFilterConfig:
287 config = FieldFilterConfig(name=field_name, model=model_name, field_name=field_name)
288 if '__enable__' in data:
289 config.enabled = bool(data.pop('__enable__'))
290 if '__logic__' in data:
291 self._set_logic_expression(config, data.pop('__logic__'))
293 for key, value in data.items():
294 config.conditions[key] = self._parse_condition(value)
296 return config
298 def _parse_filter_section(self, parts: list[str], table: Table) -> None:
299 processor_name = parts[0]
300 config = self._ensure_processor(processor_name)
302 if parts[-1] == '__filter__':
303 logic_str = table.get('__logic__')
304 if logic_str is not None:
305 self._set_logic_expression(config, self._toml_to_python(logic_str))
307 # This is replacing the whole filter section, effectively merging or resetting?
308 # Existing code seemed to merge.
309 # "fields = config.filters.setdefault(model_name, {})" -> implied merging or raw access.
310 # Here we should probably reload the filters for the affected models.
312 loaded_filters = self._load_filters(table)
313 for model_name, filter_list in loaded_filters.items():
314 # We overwrite the filters for this model if they exist, or add them.
315 # Since _load_filters creates a fresh list, we replace.
316 config.filters[model_name] = filter_list
318 config.has_filter_root = True
319 return
321 if len(parts) < 3:
322 return
324 # [Processor.__filter__.Model]
325 model_name = parts[-1]
327 # We need to parse this table as if it was inside __filter__
328 # Create a temporary table wrapper to reuse _load_filters or parse manually.
330 # Simpler: convert table to python and parse
331 model_data = self._toml_to_python(table)
333 # We need to update existing filters or create new ones for this model.
334 # If we have existing filters, we should try to preserve them?
335 # The semantics of TOML parsing usually imply "last definition wins" or "merge".
336 # But _load_filters constructs a complete list.
338 # Let's reconstruct the list for this model.
339 # But wait, if we are parsing [Processor.__filter__.Model], we have the full definition for that model here.
341 filters: list[FilterConfig] = []
342 model_filter = ModelFilterConfig(name=model_name, model=model_name)
343 if '__enable__' in model_data:
344 model_filter.enabled = bool(model_data.pop('__enable__'))
346 if '__logic__' in model_data:
347 self._set_logic_expression(model_filter, model_data.pop('__logic__'))
349 if '__conditional__' in model_data:
350 conditionals = model_data.pop('__conditional__')
351 if isinstance(conditionals, list):
352 for cond_data in conditionals:
353 filters.append(self._parse_conditional_config(model_name, cond_data))
355 for field_name, field_value in model_data.items():
356 if isinstance(field_value, dict) and not ('op' in field_value and 'value' in field_value):
357 filters.append(self._parse_field_filter_config(model_name, field_name, field_value))
358 else:
359 model_filter.conditions[field_name] = self._parse_condition(field_value)
361 filters.insert(0, model_filter)
362 config.filters[model_name] = filters
364 def add_processor(self, base_name: str, replica: str | None = None) -> None:
365 """Add a processor reference to the global processors_to_run list."""
367 ref = ProcessorRef(base_name=base_name, replica=replica)
368 target = ref.full_name
369 if target in self.globals.processors_to_run:
370 return
371 self.globals.processors_to_run.append(target)
372 if replica:
373 self._ensure_processor(base_name)
374 self._ensure_processor(target)
376 def remove_processor(self, full_name: str) -> None:
377 """Remove a processor or replica from the run list."""
379 if full_name in self.globals.processors_to_run:
380 self.globals.processors_to_run.remove(full_name)
382 def set_processors_to_run(self, processors: Iterable[str]) -> None:
383 """Overwrite the processors_to_run list."""
385 self.globals.processors_to_run = [str(p) for p in processors]
387 def set_parameter(self, processor_full_name: str, key: str, value: Any) -> None:
388 """Set a processor parameter."""
390 config = self._ensure_processor(processor_full_name)
391 if key in config.parameters:
392 config.parameters[key].value = value
393 config.parameters[key].source = ParameterSource.CONFIG
394 config.parameters[key].active_override = True
395 else:
396 config.parameters[key] = ParameterConfig(
397 name=key, value=value, source=ParameterSource.CONFIG, status=ParameterSchemaStatus.OK
398 )
399 config.parameters[key].active_override = True
401 def remove_parameter(self, processor_full_name: str, key: str) -> None:
402 """Remove a processor parameter override if present."""
404 config = self._ensure_processor(processor_full_name)
405 config.parameters.pop(key, None)
407 def clear_parameters(self, processor_full_name: str) -> None:
408 """Clear every parameter override for a processor."""
410 config = self._ensure_processor(processor_full_name)
411 config.parameters.clear()
413 def add_replica(self, base_name: str, replica: str) -> None:
414 """Create a replica entry without touching the base configuration."""
416 self._ensure_processor(base_name)
417 self._ensure_processor(f'{base_name}#{replica}')
419 def set_replica_inheritance(self, replica_full_name: str, inheritance: bool | None) -> None:
420 """Toggle the inheritance behaviour for a replica."""
422 config = self._ensure_processor(replica_full_name)
423 config.inheritance = inheritance
425 def set_processor_new_only(self, processor_full_name: str, new_only: bool | None) -> None:
426 """Explicitly set ``__new_only__`` for a processor or replica."""
428 config = self._ensure_processor(processor_full_name)
429 config.new_only = new_only
431 def set_filter_config(self, processor_full_name: str, model_name: str, config: dict[str, Any]) -> None:
432 """Replace the configuration for a given filter model."""
434 # Convert dict config back to FilterConfig objects
435 # We can reuse the logic from _parse_filter_section mostly, but we have a dict here, not TOML Table.
437 proc = self._ensure_processor(processor_full_name)
439 filters: list[FilterConfig] = []
440 model_filter = ModelFilterConfig(name=model_name, model=model_name)
442 data = config.copy() # Shallow copy to avoid mutation
444 if '__enable__' in data:
445 model_filter.enabled = bool(data.pop('__enable__'))
447 if '__logic__' in data:
448 model_filter.logic_str_original = data.pop('__logic__')
449 if model_filter.logic_str_original is not None:
450 try:
451 model_filter.logic_ast = ExprParser(model_filter.logic_str_original).parse()
452 except (mafw.db.db_filter.ParseError, ValueError):
453 model_filter.logic_ast = None
454 model_filter.logic_dirty = True
456 if '__conditional__' in data:
457 conditionals = data.pop('__conditional__')
458 if isinstance(conditionals, list):
459 for cond_data in conditionals:
460 filters.append(self._parse_conditional_config(model_name, cond_data))
462 for field_name, field_value in data.items():
463 if isinstance(field_value, dict) and not ('op' in field_value and 'value' in field_value):
464 filters.append(self._parse_field_filter_config(model_name, field_name, field_value))
465 else:
466 model_filter.conditions[field_name] = self._parse_condition(field_value)
468 filters.insert(0, model_filter)
469 proc.filters[model_name] = filters
471 def _set_logic_expression(
472 self,
473 config: ProcessorConfig | ModelFilterConfig | FieldFilterConfig,
474 logic_text: str | None,
475 *,
476 dirty: bool = False,
477 ) -> None:
478 config.logic_str_original = logic_text
479 if logic_text is None:
480 config.logic_ast = None
481 config.logic_is_valid = True
482 config.logic_dirty = dirty
483 return
484 text = str(logic_text)
485 if not text.strip():
486 config.logic_ast = None
487 config.logic_is_valid = True
488 config.logic_dirty = dirty
489 return
490 try:
491 config.logic_ast = ExprParser(text).parse()
492 config.logic_is_valid = True
493 except (mafw.db.db_filter.ParseError, ValueError):
494 config.logic_ast = None
495 config.logic_is_valid = False
496 config.logic_dirty = dirty
498 def set_filter_field(self, processor_full_name: str, model_name: str, field: str, value: Any) -> None:
499 """Set or update a single field within a filter model."""
501 proc = self._ensure_processor(processor_full_name)
502 filter_list = proc.filters.setdefault(model_name, [ModelFilterConfig(name=model_name, model=model_name)])
504 # Find the ModelFilterConfig (it should be there)
505 model_config = next((f for f in filter_list if isinstance(f, ModelFilterConfig)), None)
506 if model_config is None:
507 model_config = ModelFilterConfig(name=model_name, model=model_name)
508 filter_list.insert(0, model_config)
510 model_config.conditions[field] = self._parse_condition(value)
512 def remove_filter(self, processor_full_name: str, model_name: str) -> None:
513 """Remove a filter model definition."""
515 proc = self._ensure_processor(processor_full_name)
516 proc.filters.pop(model_name, None)
518 def set_processor_filters(
519 self, processor_full_name: str, filters: dict[str, list[FilterConfig]], logic: str | None
520 ) -> None:
521 """Update the filters and logic for a given processor."""
522 proc = self._ensure_processor(processor_full_name)
523 proc.filters = filters
524 self.set_filter_logic(processor_full_name, logic)
526 def set_filter_logic(self, processor_full_name: str, logic: str | None) -> None:
527 """Set the global ``__logic__`` string for the processor filters."""
529 proc = self._ensure_processor(processor_full_name)
530 self._set_logic_expression(proc, logic, dirty=True)
532 def set_filter_conditionals(
533 self, processor_full_name: str, model_name: str, conditionals: list[dict[str, Any]] | None
534 ) -> None:
535 """Assign ``__conditional__`` blocks to a filter model."""
537 proc = self._ensure_processor(processor_full_name)
538 filter_list = proc.filters.setdefault(model_name, [ModelFilterConfig(name=model_name, model=model_name)])
540 # Remove existing conditionals
541 proc.filters[model_name] = [f for f in filter_list if not isinstance(f, ConditionalFilterConfig)]
543 if conditionals:
544 for cond_data in conditionals:
545 proc.filters[model_name].append(self._parse_conditional_config(model_name, cond_data))
547 def set_analysis_name(self, name: str | None) -> None:
548 """Set ``analysis_name``."""
550 self.globals.analysis_name = name
552 def set_analysis_description(self, description: str | None) -> None:
553 """Set ``analysis_description``."""
555 self.globals.analysis_description = description
557 def set_new_only(self, value: bool | None) -> None:
558 """Set the top-level ``new_only`` flag."""
560 self.globals.new_only = value
562 def set_create_standard_tables(self, value: bool | None) -> None:
563 """Set ``create_standard_tables``."""
565 self.globals.create_standard_tables = value
567 def set_db_url(self, url: str | None) -> None:
568 """Override the database URL."""
570 self.db_config.url = url
571 if url is None:
572 self.db_config.attributes.pop('URL', None)
573 else:
574 self.db_config.attributes['URL'] = url
576 def set_db_pragmas(self, pragmas: dict[str, Any]) -> None:
577 """Set database pragmas."""
579 self.db_config.pragmas = dict(pragmas)
581 def set_db_authentication(self, authentication: Mapping[str, Any] | None) -> None:
582 """Set the database authentication mapping."""
584 self.db_config.authentication = dict(authentication or {})
586 def set_db_parameters(self, parameters: Mapping[str, Mapping[str, Any]] | None) -> None:
587 """Set the backend-specific database parameters mapping."""
589 normalized = {key: dict(value) for key, value in (parameters or {}).items()}
590 self.db_config.parameters = normalized
591 sqlite_pragmas = normalized.get('sqlite', {}).get('pragmas')
592 if isinstance(sqlite_pragmas, Mapping):
593 self.db_config.pragmas = dict(sqlite_pragmas)
594 else:
595 self.db_config.pragmas = {}
597 def set_default(self) -> None:
598 """Initialize globals, database, and UI defaults for a fresh builder."""
600 self.globals.analysis_name = 'analysis-name'
601 self.globals.analysis_description = 'analysis-description'
602 self.globals.new_only = True
603 self.globals.create_standard_tables = True
604 self.set_db_url('sqlite:///:memory:')
605 self.set_db_pragmas(dict(cast(dict[str, Any], DEFAULT_SQLITE_PRAGMAS)))
606 self.ui_config.interface = 'rich'
607 self.enable_db_configuration()
609 def set_db_attribute(self, key: str, value: Any | None) -> None:
610 """Store a generic key/value pair inside DBConfiguration."""
612 if value is None:
613 self.db_config.attributes.pop(key, None)
614 return
615 self.db_config.attributes[key] = value
617 def enable_db_configuration(self) -> None:
618 """Ensure the DBConfiguration section will be serialized."""
620 self.db_config.enabled = True
622 def disable_db_configuration(self) -> None:
623 """Prevent the DBConfiguration section from being emitted."""
625 self.db_config.enabled = False
627 def is_db_configuration_enabled(self) -> bool:
628 """Return whether the DBConfiguration section should be present."""
630 return self.db_config.enabled
632 def set_ui_interface(self, interface: str) -> None:
633 """Pick the interface used by ``UserInterface``."""
635 self.ui_config.interface = interface
637 def add_group(self, name: str, processors: Iterable[str], description: str | None = None) -> None:
638 """Register a processor group."""
640 self.groups[name] = GroupConfig(name=name, processors=[str(p) for p in processors], description=description)
642 def remove_group(self, name: str) -> None:
643 """Delete a group by name."""
645 self.groups.pop(name, None)
647 def list_processors(self) -> list[str]:
648 """Return every processor section name currently configured."""
650 return list(self.processors.keys())
652 def list_groups(self) -> list[str]:
653 """Return every group section name currently configured."""
655 return list(self.groups.keys())
657 @property
658 def extra_globals(self) -> dict[str, Any]:
659 """Return extra top-level globals preserved from the steering file."""
661 return self._extra_globals
663 @property
664 def document(self) -> tomlkit.TOMLDocument | None:
665 """Return the parsed TOML document this builder originated from."""
667 return self._document
669 def validate(
670 self, validation_level: ValidationLevel = ValidationLevel.SEMANTIC
671 ) -> list['mafw.mafw_errors.ValidationIssue']:
672 """Run steering validation at the requested level and report every issue."""
674 from .validation import validate as _validate # Avoid circular imports
676 return _validate(self, validation_level)
678 def to_config_dict(self) -> dict[str, Any]:
679 """Return a plain dictionary representing the steering configuration."""
681 if self._document is not None:
682 return self._document.value
683 return self.to_document().value
685 def get_processor_config(self, full_name: str) -> ProcessorConfig:
686 """Return the stored configuration for a processor or replica."""
688 return self.processors[full_name]
690 def get_group(self, name: str) -> GroupConfig:
691 """Return the stored configuration for a group section."""
693 return self.groups[name]
695 def to_document(self, *, validation_level: ValidationLevel | None = None) -> tomlkit.TOMLDocument:
696 """Serialize the builder state into a TOML document."""
698 from .serializer import serialize
700 return serialize(self, validation_level=validation_level)
702 def write(self, path: Path | str, *, validation_level: ValidationLevel | None = None) -> None:
703 """Dump the builder to disk."""
705 if isinstance(path, str):
706 path = Path(path)
707 doc = self.to_document(validation_level=validation_level)
708 with path.open('w', encoding='utf-8') as handle:
709 tomlkit.dump(doc, handle)
711 def _ensure_processor(self, name: str) -> ProcessorConfig:
712 if name not in self.processors:
713 self.processors[name] = ProcessorConfig(name=name)
714 return self.processors[name]