Add support for subscribing to specific config keys

This is useful for actors to be able to subscribe to a particular key
with the actor configuration, avoiding spurious updates and noise in
the received configuration.

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

Author: llucax

src/frequenz/sdk/config/_manager.py

@@ -8,7 +8,7 @@
 import pathlib
 from collections.abc import Mapping, MutableMapping, Sequence, Set
 from datetime import timedelta
-from typing import Any
+from typing import Any, assert_never, overload
 
 from frequenz.channels import Broadcast, Receiver
 from frequenz.channels.file_watcher import EventType
@@ -120,12 +120,30 @@ def __str__(self) -> str:
         """Return a string representation of this config manager."""
         return f"{type(self).__name__}[{self._name}]"
 
+    @overload
+    async def new_receiver(
+        self,
+        *,
+        wait_for_first: bool = True,
+        skip_unchanged: bool = True,
+    ) -> Receiver[Mapping[str, Any]]: ...
+
+    @overload
+    async def new_receiver(
+        self,
+        *,
+        wait_for_first: bool = True,
+        skip_unchanged: bool = True,
+        key: str,
+    ) -> Receiver[Mapping[str, Any] | None]: ...
+
     # The noqa DOC502 is needed because we raise TimeoutError indirectly.
     async def new_receiver(  # pylint: disable=too-many-arguments # noqa: DOC502
         self,
         *,
         wait_for_first: bool = False,
         skip_unchanged: bool = True,
+        key: str | None = None,
     ) -> Receiver[Mapping[str, Any] | None]:
         """Create a new receiver for the configuration.
 
@@ -143,6 +161,13 @@ async def new_receiver(  # pylint: disable=too-many-arguments # noqa: DOC502
         The comparison is done using the *raw* `dict` to determine if the configuration
         has changed.
 
+        ### Filtering
+
+        The configuration can be filtered by a `key`. If the `key` is `None`, the
+        receiver will receive the full configuration, otherwise only the part of the
+        configuration under the specified key is received, or `None` if the key is not
+        found.
+
         ### Waiting for the first configuration
 
         If `wait_for_first` is `True`, the receiver will wait for the first
@@ -168,6 +193,9 @@ async def new_receiver(  # pylint: disable=too-many-arguments # noqa: DOC502
                 [`consume()`][frequenz.channels.Receiver.consume] on the receiver.
             skip_unchanged: Whether to skip sending the configuration if it hasn't
                 changed compared to the last one received.
+            key: The key to filter the configuration. A nested key can be specified by
+                using a dot (`.`) as separator. For example "key.subkey" will get only
+                `config[key][subkey]` If `None`, all configurations will be received.
 
         Returns:
             The receiver for the configuration.
@@ -176,11 +204,18 @@ async def new_receiver(  # pylint: disable=too-many-arguments # noqa: DOC502
             asyncio.TimeoutError: If `wait_for_first` is `True` and the first
                 configuration can't be received in time.
         """
-        recv_name = f"{self._name}_receiver"
+        recv_name = (
+            f"{self._name}_receiver" if key is None else f"{self._name}_receiver_{key}"
+        )
         receiver = self._config_channel.new_receiver(name=recv_name, limit=1)
 
         if skip_unchanged:
-            receiver = receiver.filter(self._config_has_changed)
+            receiver = receiver.filter(
+                lambda config: self._config_has_changed(config, key)
+            )
+
+        if key is not None:
+            receiver = receiver.map(lambda config: config.get(key))
 
         if wait_for_first:
             async with asyncio.timeout(self.wait_for_first_timeout.total_seconds()):
@@ -191,11 +226,31 @@ async def new_receiver(  # pylint: disable=too-many-arguments # noqa: DOC502
     def _config_has_changed(
         self,
         new_config: Mapping[str, Any] | None,
+        key: str | None,
     ) -> bool:
         """Return whether the configuration has changed."""
         old_config = self._last_raw_config
-        if new_config != old_config:
-            return True
-        _logger.info("Configuration has not changed, skipping update")
-        _logger.debug("Old configuration being kept: %r", old_config)
-        return False
+        if key is None:
+            has_changed = new_config != old_config
+        else:
+            match (new_config, old_config):
+                case (None, None):
+                    has_changed = False
+                case (None, old):
+                    has_changed = key in old
+                case (new, None):
+                    has_changed = key in new
+                case (new, old):
+                    has_changed = new.get(key) != old.get(key)
+                case unexpected:
+                    assert_never(unexpected)
+
+        if not has_changed:
+            key_str = f" for key '{key}'" if key else ""
+            _logger.info(
+                "Configuration%s has not changed, skipping update",
+                key_str,
+            )
+            _logger.debug("Old configuration%s being kept: %r", key_str, old_config)
+
+        return has_changed