Validate and load configurations using a config dataclass

This commit adds support for validating configurations with a schema.
This is useful to ensure the configuration is correct and to receive it
as an instance of a dataclass.

The `new_receiver` method now requires a `config_class` argument that is
used to validate the configuration. If the configuration doesn't pass
the validation, it will be ignored and an error will be logged.

The schema is expected to be a dataclass, which is used to create a
marshmallow schema to validate the configuration. To customize the
schema, you can use `marshmallow_dataclass` to specify extra metadata.

Validation errors are logged and the `ValidationError` instance is sent
to the receiver.

Signed-off-by: Leandro Lucarella <[email protected]>

Author: llucax

src/frequenz/sdk/config/_manager.py

@@ -11,10 +11,12 @@
 
 from frequenz.channels import Broadcast, Receiver
 from frequenz.channels.experimental import WithPrevious
+from marshmallow import Schema, ValidationError
 from typing_extensions import override
 
 from ..actor._background_service import BackgroundService
 from ._managing_actor import ConfigManagingActor
+from ._util import DataclassT, load_config
 
 _logger = logging.getLogger(__name__)
 
@@ -124,18 +126,21 @@ def __repr__(self) -> str:
         """Return a string representation of this config manager."""
         return f"config_channel={self.config_channel!r}, " f"actor={self.actor!r}>"
 
-    def new_receiver(
+    def new_receiver(  # pylint: disable=too-many-arguments
         self,
         # This is tricky, because a str is also a Sequence[str], if we would use only
         # Sequence[str], then a regular string would also be accepted and taken as
         # a sequence, like "key" -> ["k", "e", "y"]. We should never remove the str from
         # the allowed types without changing Sequence[str] to something more specific,
         # like list[str] or tuple[str] (but both have their own problems).
         key: str | Sequence[str],
+        config_class: type[DataclassT],
         /,
         *,
         skip_unchanged: bool = True,
-    ) -> Receiver[Mapping[str, Any] | InvalidValueForKeyError | None]:
+        base_schema: type[Schema] | None = None,
+        marshmallow_load_kwargs: dict[str, Any] | None = None,
+    ) -> Receiver[DataclassT | Exception | None]:
         """Create a new receiver for receiving the configuration for a particular key.
 
         This method has a lot of features and functionalities to make it easier to
@@ -157,6 +162,21 @@ def new_receiver(
         will be logged and a [`frequenz.sdk.config.InvalidValueForKeyError`][] instance
         will be sent to the receiver.
 
+        ### Schema validation
+
+        The raw configuration received as a `Mapping` will be validated and loaded to
+        as a `config_class`. The `config_class` class is expected to be
+        a [`dataclasses.dataclass`][], which is used to create
+        a [`marshmallow.Schema`][] via the [`marshmallow_dataclass.class_schema`][]
+        function.
+
+        This means you can customize the schema derived from the configuration
+        dataclass using [`marshmallow_dataclass`][] to specify extra validation and
+        options via field metadata.
+
+        Additional arguments can be passed to [`marshmallow.Schema.load`][] using
+        the `marshmallow_load_kwargs` keyword arguments.
+
         ### Skipping superfluous updates
 
         If there is a burst of configuration updates, the receiver will only receive the
@@ -167,6 +187,18 @@ def new_receiver(
         The comparison is done using the *raw* `dict` to determine if the configuration
         has changed.
 
+        ### Error handling
+
+        The value under `key` must be another mapping, otherwise an error
+        will be logged and a [`frequenz.sdk.config.InvalidValueForKeyError`][] instance
+        will be sent to the receiver.
+
+        Configurations that don't pass the validation will be logged as an error and
+        the [`ValidationError`][marshmallow.ValidationError] sent to the receiver.
+
+        Any other unexpected error raised during the configuration loading will be
+        logged as an error and the error instance sent to the receiver.
+
         Example:
             ```python
             # TODO: Add Example
@@ -175,44 +207,49 @@ def new_receiver(
         Args:
             key: The configuration key to be read by the receiver. If a sequence of
                 strings is used, it is used as a sub-key.
+            config_class: The class object to use to instantiate a configuration. The
+                configuration will be validated against this type too using
+                [`marshmallow_dataclass`][].
             skip_unchanged: Whether to skip sending the configuration if it hasn't
                 changed compared to the last one received.
+            base_schema: An optional class to be used as a base schema for the
+                configuration class. This allow using custom fields for example. Will be
+                passed to [`marshmallow_dataclass.class_schema`][].
+            marshmallow_load_kwargs: Additional arguments to be passed to
+                [`marshmallow.Schema.load`][].
 
         Returns:
             The receiver for the configuration.
         """
-        recv_name = key if isinstance(key, str) else ":".join(key)
-        receiver = self.config_channel.new_receiver(name=recv_name, limit=1)
-
-        def _get_key_or_error(
-            config: Mapping[str, Any]
-        ) -> Mapping[str, Any] | InvalidValueForKeyError | None:
-            try:
-                return _get_key(config, key)
-            except InvalidValueForKeyError as error:
-                return error
-
-        key_receiver = receiver.map(_get_key_or_error)
+        receiver = self.config_channel.new_receiver(name=f"{self}:{key}", limit=1).map(
+            lambda config: _load_config_with_logging_and_errors(
+                config,
+                config_class,
+                key=key,
+                base_schema=base_schema,
+                marshmallow_load_kwargs=marshmallow_load_kwargs,
+            )
+        )
 
         if skip_unchanged:
             # For some reason the type argument for WithPrevious is not inferred
             # correctly, so we need to specify it explicitly.
-            return key_receiver.filter(
-                WithPrevious[Mapping[str, Any] | InvalidValueForKeyError | None](
+            return receiver.filter(
+                WithPrevious[DataclassT | Exception | None](
                     lambda old, new: _not_equal_with_logging(
                         key=key, old_value=old, new_value=new
                     )
                 )
             )
 
-        return key_receiver
+        return receiver
 
 
 def _not_equal_with_logging(
     *,
     key: str | Sequence[str],
-    old_value: Mapping[str, Any] | InvalidValueForKeyError | None,
-    new_value: Mapping[str, Any] | InvalidValueForKeyError | None,
+    old_value: DataclassT | Exception | None,
+    new_value: DataclassT | Exception | None,
 ) -> bool:
     """Return whether the two mappings are not equal, logging if they are the same."""
     if old_value == new_value:
@@ -234,6 +271,47 @@ def _not_equal_with_logging(
     return True
 
 
+def _load_config_with_logging_and_errors(
+    config: Mapping[str, Any],
+    config_class: type[DataclassT],
+    *,
+    key: str | Sequence[str],
+    base_schema: type[Schema] | None = None,
+    marshmallow_load_kwargs: dict[str, Any] | None = None,
+) -> DataclassT | Exception | None:
+    """Load the configuration for the specified key, logging errors and returning them."""
+    try:
+        sub_config = _get_key(config, key)
+        if sub_config is None:
+            _logger.debug("Configuration key %r not found, sending None", key)
+            return None
+
+        loaded_config = load_config(
+            config_class,
+            sub_config,
+            base_schema=base_schema,
+            marshmallow_load_kwargs=marshmallow_load_kwargs,
+        )
+        _logger.debug("Received new configuration: %s", loaded_config)
+        return loaded_config
+    except InvalidValueForKeyError as error:
+        if len(key) > 1 and key != error.key:
+            _logger.error("Error when looking for sub-key %r: %s", key, error)
+        else:
+            _logger.error(str(error))
+        return error
+    except ValidationError as error:
+        _logger.error("The configuration for key %r is invalid: %s", key, error)
+        return error
+    except Exception as error:  # pylint: disable=broad-except
+        _logger.exception(
+            "An unexpected error occurred while loading the configuration for key %r: %s",
+            key,
+            error,
+        )
+        return error
+
+
 def _get_key(
     config: Mapping[str, Any],
     # This is tricky, because a str is also a Sequence[str], if we would use only