From 5bb8391633c6178e2d6407b84e1af3b8b989b550 Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Tue, 3 Mar 2026 11:32:43 +0100 Subject: [PATCH 1/2] Add experimental TOML config generator from marshmallow schemas Walk marshmallow schemas (typically from marshmallow_dataclass) and emit TOML text with default values and field descriptions as comments. Handles nested sections, quantity fields as string leaves, optional sections (commented out), enums, timedeltas, and infinity defaults. Extends the existing experimental marshmallow integration with a generate_toml_from_schema() entry point and a CommentStyle enum. Signed-off-by: Mathias L. Baumann --- .../quantities/experimental/toml_generator.py | 363 +++++++++++++++++ tests/experimental/test_toml_generator.py | 366 ++++++++++++++++++ 2 files changed, 729 insertions(+) create mode 100644 src/frequenz/quantities/experimental/toml_generator.py create mode 100644 tests/experimental/test_toml_generator.py diff --git a/src/frequenz/quantities/experimental/toml_generator.py b/src/frequenz/quantities/experimental/toml_generator.py new file mode 100644 index 0000000..ade0192 --- /dev/null +++ b/src/frequenz/quantities/experimental/toml_generator.py @@ -0,0 +1,363 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Generate TOML configuration from marshmallow schemas. + +This module walks marshmallow schemas (typically created via +`marshmallow_dataclass.class_schema`) and emits TOML text with default values +and field descriptions as comments. + +The main entry point is +[`generate_toml_from_schema`][frequenz.quantities.experimental.toml_generator.generate_toml_from_schema], +which accepts a marshmallow schema and a TOML section name and returns the +rendered TOML string. + +Danger: + This module contains experimental features for which the API is not yet stable. + + Any module or class in this package may be removed or changed in a future release, + even in minor or patch releases. + +Example: + ```python + from dataclasses import dataclass, field + from marshmallow_dataclass import class_schema + from frequenz.quantities import Power + from frequenz.quantities.experimental.marshmallow import QuantitySchema + from frequenz.quantities.experimental.toml_generator import ( + generate_toml_from_schema, + ) + + @dataclass + class MyConfig: + threshold: float = field( + default=0.5, + metadata={"metadata": {"description": "Activation threshold."}}, + ) + name: str = field( + default="default", + metadata={"metadata": {"description": "Instance name."}}, + ) + max_power: Power = field( + default_factory=lambda: Power.from_kilowatts(10.0), + metadata={"metadata": {"description": "Maximum power output."}}, + ) + + schema = class_schema(MyConfig, base_schema=QuantitySchema)() + print(generate_toml_from_schema(schema, "my_config")) + ``` +""" + +from __future__ import annotations + +import enum +from datetime import timedelta +from typing import Any + +import marshmallow + +# Maximum line length used by the SHORT comment style to decide inline vs above. +_COMMENT_WRAP = 100 + + +class CommentStyle(enum.Enum): + """Controls where field descriptions are placed in the generated TOML. + + Members: + AUTO: Inline when a usable default exists, above otherwise (default). + ABOVE: Always place the comment on the line(s) above the key. + INLINE: Always place the comment on the same line as the key. + SHORT: Inline when the resulting line fits within the wrap limit, + above otherwise. + """ + + AUTO = "auto" + ABOVE = "above" + INLINE = "inline" + SHORT = "short" + + +def _resolve_nested(field: marshmallow.fields.Nested) -> marshmallow.Schema | None: + """Resolve a Nested field to its schema instance, or None.""" + nested: Any = field.nested + if callable(nested) and not isinstance(nested, type): + try: + nested = nested() + except Exception: # pylint: disable=broad-except + return None + if isinstance(nested, type): + try: + nested = nested() + except Exception: # pylint: disable=broad-except + return None + if isinstance(nested, marshmallow.Schema): + return nested + return None + + +def _resolve_default(default: Any) -> Any: + """Resolve a default value, calling it if it's a callable.""" + if default is marshmallow.missing: + return marshmallow.missing + if callable(default) and not isinstance(default, type): + try: + return default() + except Exception: # pylint: disable=broad-except + return marshmallow.missing + return default + + +def _is_quantity_schema(schema: marshmallow.Schema | None) -> bool: + """Check whether a schema wraps a Quantity type (Power, Percentage, etc.).""" + if schema is None or schema.fields: + return False + try: + from .marshmallow import ( # pylint: disable=import-outside-toplevel + QUANTITY_FIELD_CLASSES, + ) + + return type(schema).__name__ in {k.__name__ for k in QUANTITY_FIELD_CLASSES} + except ImportError: + return False + + +def _has_usable_default(default: Any) -> bool: + """Return True if *default* is a concrete value we can emit.""" + if default is None or default is marshmallow.missing: + return False + if isinstance(default, type): + return False + if isinstance(default, float) and ( + default == float("inf") or default == float("-inf") + ): + return False + return True + + +def _format_value(value: Any, *, by_value: bool = True) -> str: + """Format a Python value as a TOML literal.""" + if isinstance(value, bool): + return "true" if value else "false" + if isinstance(value, str): + return f'"{value}"' + if isinstance(value, timedelta): + return str(int(value.total_seconds())) + if isinstance(value, enum.Enum): + return f'"{value.value if by_value else value.name}"' + if isinstance(value, (int, float)): + return str(value) + if isinstance(value, (list, tuple, set)): + if not value: + return "[]" + items = ", ".join(_format_value(v) for v in value) + return f"[{items}]" + # Quantity objects — use their human-readable string. + if hasattr(value, "base_value"): + return f'"{value}"' + return repr(value) + + +def _first_line(text: str) -> str: + """Return the first line of *text*.""" + return text.split("\n", 1)[0].strip() + + +def _first_sentence(text: str) -> str: + """Return the first sentence of *text* (up to the first '.', '!', or '?').""" + for i, ch in enumerate(text): + if ch in ".!?": + return text[: i + 1].strip() + return text.strip() + + +def _above_comment(desc: str, *, first_sentence_only: bool) -> str: + """Format *desc* as a ``# …`` comment line to appear above a key.""" + text = _first_sentence(desc) if first_sentence_only else _first_line(desc) + return f"# {text}" if text else "" + + +def _comment_out(text: str) -> str: + """Prefix every non-empty, non-comment line with ``# ``.""" + out: list[str] = [] + for line in text.split("\n"): + if line and not line.startswith("#"): + out.append(f"# {line}") + else: + out.append(line) + return "\n".join(out) + + +def _emit_leaf( # pylint: disable=too-many-arguments + key: str, + field: marshmallow.fields.Field[Any], + default: Any, + desc: str, + *, + style: CommentStyle = CommentStyle.AUTO, + first_sentence_only: bool = False, +) -> str: + """Return one or two TOML lines for a leaf field. + + Returns a single line for inline styles, or a comment line followed by + the key-value line when the comment is placed above. + """ + by_value = ( + getattr(field, "by_value", True) + if isinstance(field, marshmallow.fields.Enum) + else True + ) + has_default = _has_usable_default(default) + value_str = _format_value(default, by_value=by_value) if has_default else None + + # Build the key=value part (without any comment). + if has_default: + kv = f"{key} = {value_str}" + else: + kv = f"# {key} =" + + if not desc: + return kv + + inline_comment = f" # {_first_line(desc)}" + above_comment = _above_comment(desc, first_sentence_only=first_sentence_only) + + def _with_inline() -> str: + return f"{kv}{inline_comment}" + + def _with_above() -> str: + if above_comment: + return f"{above_comment}\n{kv}" + return kv + + if style == CommentStyle.INLINE: + return _with_inline() + if style == CommentStyle.ABOVE: + return _with_above() + if style == CommentStyle.AUTO: + # Inline when we have a default, above when we don't. + return _with_inline() if has_default else _with_above() + # SHORT: inline if the full line fits within the wrap limit, above otherwise. + inline_line = _with_inline() + if len(inline_line) <= _COMMENT_WRAP: + return inline_line + return _with_above() + + +# Each entry is (section_path, content_lines_str). +_Section = tuple[str, str] + + +def _collect_sections( + schema: marshmallow.Schema, + section_path: str, + *, + style: CommentStyle = CommentStyle.AUTO, + first_sentence_only: bool = False, +) -> list[_Section]: + """Walk a schema and return ``(section_path, body)`` pairs for TOML output.""" + sections: list[_Section] = [] + inline_lines: list[str] = [] + + for name, field in schema.fields.items(): + key = field.data_key or name + desc = field.metadata.get("description", "") + default = _resolve_default(field.load_default) + + # --- Nested fields may become sub-sections or leaf values --- + if isinstance(field, marshmallow.fields.Nested): + nested_schema = _resolve_nested(field) + + if _is_quantity_schema(nested_schema): + # Quantity → treat as a leaf value. + inline_lines.append( + _emit_leaf( + key, + field, + default, + desc, + style=style, + first_sentence_only=first_sentence_only, + ) + ) + elif nested_schema is not None and nested_schema.fields: + sub = _collect_sections( + nested_schema, + f"{section_path}.{key}", + style=style, + first_sentence_only=first_sentence_only, + ) + if default is None: + # Optional section → comment everything out. + sections.extend((p, _comment_out(c)) for p, c in sub) + else: + sections.extend(sub) + else: + # Unresolvable nested (e.g. forward-ref string) → leaf. + inline_lines.append( + _emit_leaf( + key, + field, + default, + desc, + style=style, + first_sentence_only=first_sentence_only, + ) + ) + continue + + # --- All other field types are leaves --- + inline_lines.append( + _emit_leaf( + key, + field, + default, + desc, + style=style, + first_sentence_only=first_sentence_only, + ) + ) + + body = "\n".join(inline_lines) + if body.strip(): + sections.insert(0, (section_path, body)) + elif not sections: + sections.append((section_path, "")) + + return sections + + +def generate_toml_from_schema( + schema: marshmallow.Schema, + section: str, + *, + style: CommentStyle = CommentStyle.AUTO, + first_sentence_only: bool = False, +) -> str: + """Generate TOML text from a marshmallow schema. + + Walks the schema's fields recursively, emitting TOML sections with + default values and field descriptions as comments. + + Nested dataclass fields become ``[section.subsection]`` headers. + Quantity fields (Power, Percentage, etc.) are rendered as string + leaf values. Optional nested sections (where the default is ``None``) + are emitted fully commented out. + + Args: + schema: A marshmallow Schema instance to walk. + section: The top-level TOML section name (e.g. ``"my_actor"``). + style: Where to place field descriptions relative to the key. + first_sentence_only: When placing comments above a key, truncate + to the first sentence rather than the first line. + + Returns: + The generated TOML text as a string (without trailing newline). + """ + lines: list[str] = [] + for section_path, content in _collect_sections( + schema, section, style=style, first_sentence_only=first_sentence_only + ): + lines.append(f"[{section_path}]") + if content.strip(): + lines.append(content) + return "\n".join(lines) diff --git a/tests/experimental/test_toml_generator.py b/tests/experimental/test_toml_generator.py new file mode 100644 index 0000000..30edd45 --- /dev/null +++ b/tests/experimental/test_toml_generator.py @@ -0,0 +1,366 @@ +# License: MIT +# Copyright © 2025 Frequenz Energy-as-a-Service GmbH + +"""Test TOML config generation from marshmallow schemas.""" + +from __future__ import annotations + +import enum +from dataclasses import dataclass, field +from datetime import timedelta +from typing import Optional + +from marshmallow_dataclass import class_schema + +from frequenz.quantities import Percentage, Power +from frequenz.quantities.experimental.marshmallow import QuantitySchema +from frequenz.quantities.experimental.toml_generator import ( + CommentStyle, + generate_toml_from_schema, +) + +# ---- Test dataclasses ---- + + +@dataclass +class SimpleConfig: + """A simple config with basic leaf fields.""" + + name: str = field( + default="hello", + metadata={"metadata": {"description": "The instance name."}}, + ) + count: int = field( + default=42, + metadata={"metadata": {"description": "Number of retries."}}, + ) + enabled: bool = field( + default=True, + metadata={"metadata": {"description": "Whether the feature is enabled."}}, + ) + ratio: float = field( + default=0.75, + metadata={"metadata": {"description": "Ratio value."}}, + ) + + +@dataclass +class NoDefaultConfig: + """Config with a field that has no default.""" + + required_field: str = field( + metadata={"metadata": {"description": "This field is required."}}, + ) + + +@dataclass +class NestedChild: + """A child config section.""" + + timeout: int = field( + default=30, + metadata={"metadata": {"description": "Timeout in seconds."}}, + ) + verbose: bool = field( + default=False, + metadata={"metadata": {"description": "Enable verbose logging."}}, + ) + + +@dataclass +class NestedParent: + """A parent config with a nested section.""" + + label: str = field( + default="main", + metadata={"metadata": {"description": "The label."}}, + ) + child: NestedChild = field(default_factory=NestedChild) + + +@dataclass +class OptionalNestedParent: + """A parent config with an optional nested section.""" + + label: str = field( + default="main", + metadata={"metadata": {"description": "The label."}}, + ) + child: Optional[NestedChild] = field( # noqa: UP007 + default=None, + ) + + +class Color(enum.Enum): + """A color enum.""" + + RED = "red" + GREEN = "green" + BLUE = "blue" + + +@dataclass +class EnumConfig: + """Config with an enum field.""" + + color: Color = field( + default=Color.GREEN, + metadata={"metadata": {"description": "The primary color."}}, + ) + + +@dataclass +class TimedeltaConfig: + """Config with a timedelta field.""" + + interval: timedelta = field( + default_factory=lambda: timedelta(seconds=60), + metadata={"metadata": {"description": "Polling interval."}}, + ) + + +@dataclass +class ListConfig: + """Config with a list field.""" + + tags: list[str] = field( + default_factory=list, + metadata={"metadata": {"description": "A list of tags."}}, + ) + + +@dataclass +class NoDescConfig: + """Config with a field that has no description.""" + + value: int = field(default=10) + + +@dataclass +class QuantityConfig: + """Config with quantity fields.""" + + max_power: Power = field( + default_factory=lambda: Power.from_kilowatts(10.0), + metadata={"metadata": {"description": "Maximum power output."}}, + ) + threshold: Percentage = field( + default_factory=lambda: Percentage.from_percent(80.0), + metadata={"metadata": {"description": "Activation threshold."}}, + ) + label: str = field( + default="test", + metadata={"metadata": {"description": "A label."}}, + ) + + +@dataclass +class InfDefaultConfig: + """Config with infinity defaults (should be treated as no-default).""" + + upper: float = field( + default=float("inf"), + metadata={"metadata": {"description": "Upper bound."}}, + ) + lower: float = field( + default=float("-inf"), + metadata={"metadata": {"description": "Lower bound."}}, + ) + + +# ---- Tests ---- + + +def test_simple_leaf_fields() -> None: + """Test generation of simple leaf fields with defaults and descriptions.""" + schema = class_schema(SimpleConfig)() + result = generate_toml_from_schema(schema, "simple") + + assert "[simple]" in result + assert 'name = "hello" # The instance name.' in result + assert "count = 42 # Number of retries." in result + assert "enabled = true # Whether the feature is enabled." in result + assert "ratio = 0.75 # Ratio value." in result + + +def test_no_default_field() -> None: + """Test that fields without defaults are commented out.""" + schema = class_schema(NoDefaultConfig)() + result = generate_toml_from_schema(schema, "section") + + assert "# This field is required." in result + assert "# required_field =" in result + + +def test_nested_section() -> None: + """Test that nested dataclasses become sub-sections.""" + schema = class_schema(NestedParent)() + result = generate_toml_from_schema(schema, "parent") + + assert "[parent]" in result + assert 'label = "main"' in result + assert "[parent.child]" in result + assert "timeout = 30" in result + assert "verbose = false" in result + + +def test_optional_nested_commented_out() -> None: + """Test that optional nested sections (default=None) are fully commented out.""" + schema = class_schema(OptionalNestedParent)() + result = generate_toml_from_schema(schema, "parent") + + assert "[parent]" in result + assert 'label = "main"' in result + # The child section header should still be present but the body commented out + assert "[parent.child]" in result + assert "# timeout = 30" in result + assert "# verbose = false" in result + + +def test_enum_field() -> None: + """Test enum fields render as a quoted string.""" + schema = class_schema(EnumConfig)() + result = generate_toml_from_schema(schema, "cfg") + + # marshmallow serializes enums by name by default + assert 'color = "GREEN"' in result + + +def test_timedelta_field() -> None: + """Test timedelta fields render as integer seconds.""" + schema = class_schema(TimedeltaConfig)() + result = generate_toml_from_schema(schema, "cfg") + + assert "interval = 60" in result + + +def test_empty_list_field() -> None: + """Test that empty list defaults are treated as no-default (commented out).""" + schema = class_schema(ListConfig)() + result = generate_toml_from_schema(schema, "cfg") + + # Empty list is falsy → _has_usable_default returns False → commented out + assert "# tags =" in result + + +def test_no_description() -> None: + """Test fields without descriptions produce no comment.""" + schema = class_schema(NoDescConfig)() + result = generate_toml_from_schema(schema, "cfg") + + assert "value = 10" in result + assert "#" not in result.split("value = 10")[1].split("\n")[0] + + +def test_comment_style_above() -> None: + """Test ABOVE comment style always puts comments above.""" + schema = class_schema(SimpleConfig)() + result = generate_toml_from_schema(schema, "s", style=CommentStyle.ABOVE) + + lines = result.split("\n") + # Find the "name" key line + for i, line in enumerate(lines): + if line.startswith('name = "hello"'): + assert lines[i - 1] == "# The instance name." + # No inline comment + assert "#" not in line + break + else: + raise AssertionError("name key not found") + + +def test_comment_style_inline() -> None: + """Test INLINE comment style always puts comments inline.""" + schema = class_schema(NoDefaultConfig)() + result = generate_toml_from_schema(schema, "s", style=CommentStyle.INLINE) + + # Even fields without defaults get inline comments + assert "# required_field = # This field is required." in result + + +def test_comment_style_short() -> None: + """Test SHORT comment style uses inline when line fits.""" + schema = class_schema(SimpleConfig)() + result = generate_toml_from_schema(schema, "s", style=CommentStyle.SHORT) + + # Short enough → inline + assert 'name = "hello" # The instance name.' in result + + +def test_quantity_fields_as_leaves() -> None: + """Test that Quantity fields are rendered as leaf string values.""" + schema = class_schema(QuantityConfig, base_schema=QuantitySchema)() + result = generate_toml_from_schema(schema, "actor") + + assert "[actor]" in result + # Quantities should be rendered as string leaves, not sub-sections + assert "max_power" in result + assert "threshold" in result + assert 'label = "test"' in result + # Should NOT have [actor.max_power] as a sub-section + assert "[actor.max_power]" not in result + assert "[actor.threshold]" not in result + + +def test_infinity_defaults_treated_as_no_default() -> None: + """Test that +/-inf defaults are treated as no-default (commented out).""" + schema = class_schema(InfDefaultConfig)() + result = generate_toml_from_schema(schema, "cfg") + + assert "# upper =" in result + assert "# lower =" in result + + +def test_first_sentence_only() -> None: + """Test first_sentence_only truncation of descriptions.""" + + @dataclass + class LongDescConfig: + """Config with a long multi-sentence description.""" + + value: int = field( + default=5, + metadata={ + "metadata": { + "description": "The main value. Used for calculations. Do not change." + } + }, + ) + + schema = class_schema(LongDescConfig)() + + # Without first_sentence_only — should show full first line + result_full = generate_toml_from_schema(schema, "cfg") + assert "The main value. Used for calculations. Do not change." in result_full + + # With first_sentence_only on ABOVE style — should truncate + result_short = generate_toml_from_schema( + schema, "cfg", style=CommentStyle.ABOVE, first_sentence_only=True + ) + assert "# The main value." in result_short + assert "Used for calculations" not in result_short + + +def test_multiple_sections_output_format() -> None: + """Test that the overall output format has proper section headers.""" + schema = class_schema(NestedParent)() + result = generate_toml_from_schema(schema, "my_actor") + + lines = result.split("\n") + # First non-empty line should be the section header + assert lines[0] == "[my_actor]" + # Should contain the child section somewhere + assert any(line == "[my_actor.child]" for line in lines) + + +def test_empty_schema() -> None: + """Test that an empty schema produces a section header with empty body.""" + + @dataclass + class EmptyConfig: + """Config with no fields.""" + + schema = class_schema(EmptyConfig)() + result = generate_toml_from_schema(schema, "empty") + + assert "[empty]" in result From e08ebd786e45d47ff34e6afa353b14e4be32363c Mon Sep 17 00:00:00 2001 From: "Mathias L. Baumann" Date: Tue, 3 Mar 2026 11:51:50 +0100 Subject: [PATCH 2/2] docs: Add release notes for toml_generator module Signed-off-by: Mathias L. Baumann --- RELEASE_NOTES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index 6a5f7fe..a2eedb6 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -10,7 +10,7 @@ ## New Features - +- Add `frequenz.quantities.experimental.toml_generator` module with `generate_toml_from_schema()` and `CommentStyle` for generating documented TOML configuration files from marshmallow schemas. ## Bug Fixes