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 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