WIP

Author: llucax

src/frequenz/sdk/config/__init__.py

@@ -1,20 +1,88 @@
 # License: MIT
 # Copyright © 2024 Frequenz Energy-as-a-Service GmbH
 
-"""Read and update config variables."""
+"""Configuration management.
+
+Example: App configuring the global config manager.
+    ```python
+    import asyncio
+    import dataclasses
+    import sys
+
+    import marshmallow
+
+    from frequenz.channels import select, selected_from
+    from frequenz.sdk.actor import Actor
+    from frequenz.sdk.config import (
+        initialize_config,
+        config_manager,
+        LoggingConfigUpdatingActor,
+        ConfigManager,
+    )
+
+    @dataclasses.dataclass
+    class ActorConfig:
+        name: str
+
+    class MyActor(Actor):
+        def __init__(self, config: ActorConfig) -> None:
+            self._config = config
+
+        async def _run(self) -> None:
+            receiver = ...
+            config_receiver = await config_manager().new_receiver(schema=ActorConfig)
+
+            async for selected in select(receiver, config_receiver):
+                if selected_from(selected, receiver):
+                    ...
+                elif selected_from(selected, config_receiver):
+                    self._config = selected.message
+                    # Restart whatever is needed after a config update
+
+
+    @dataclasses.dataclass
+    class AppConfig:
+        positive_int: int = dataclasses.field(
+            default=42,
+            metadata={"validate": marshmallow.validate.Range(min=0)},
+        )
+        my_actor: ActorConfig | None = None
+        logging: LoggingConfig = LoggingConfig()
+
+    async def main() -> None:
+        config_manager = initialize_config_manager(["config.toml"])
+        try:
+            # Receive the first configuration
+            initial_config = await config_manager.new_receiver(schema=AppConfig,
+                                                                  wait_for_first=True)
+        except asyncio.TimeoutError:
+            print("No configuration received in time")
+            sys.exit(1)
+
+        actor = MyActor(ActorConfig(name=initial_config.my_actor))
+        actor.start()
+        await actor
+    ```
+"""
 
 from ._actor import ConfigManagingActor
+from ._global import config_manager, initialize_config, shutdown_config_manager
 from ._logging_config_updater import (
     LoggerConfig,
     LoggingConfig,
     LoggingConfigUpdatingActor,
 )
+from ._manager import ConfigManager
 from ._util import load_config
 
 __all__ = [
+    "ConfigManager",
     "ConfigManagingActor",
     "LoggingConfig",
     "LoggerConfig",
     "LoggingConfigUpdatingActor",
     "load_config",
+    "config_manager",
+    "initialize_config",
+    "shutdown_config_manager",
 ]

src/frequenz/sdk/config/_global.py

@@ -0,0 +1,153 @@
+# License: MIT
+# Copyright © 2024 Frequenz Energy-as-a-Service GmbH
+
+"""Global config manager."""
+
+import asyncio
+import logging
+import pathlib
+from collections import abc
+from collections.abc import Sequence
+from datetime import timedelta
+
+from frequenz.channels.file_watcher import EventType
+
+from ._manager import ConfigManager
+
+_logger = logging.getLogger(__name__)
+
+# pylint: disable=global-statement
+_CONFIG_MANAGER: ConfigManager | None = None
+"""Global instance of the ConfigManagingActor.
+
+This is the only instance of the ConfigManagingActor that should be used in the
+entire application. It is created lazily on the first access and should be
+accessed via the `get_config_manager` function.
+"""
+
+
+def initialize_config(  # pylint: disable=too-many-arguments
+    config_paths: Sequence[pathlib.Path],
+    /,
+    *,
+    event_types: abc.Set[EventType] = frozenset(EventType),
+    force_polling: bool = True,
+    name: str = "global",
+    polling_interval: timedelta = timedelta(seconds=5),
+    wait_for_first_timeout: timedelta = timedelta(seconds=5),
+) -> ConfigManager:
+    """Initialize the singleton instance of the ConfigManagingActor.
+
+    Args:
+        config_paths: Paths to the TOML configuration files.
+        event_types: The set of event types to monitor.
+        force_polling: Whether to force file polling to check for changes.
+        name: The name of the config manager.
+        polling_interval: The interval to poll for changes. Only relevant if
+            polling is enabled.
+        wait_for_first_timeout: The timeout to use when waiting for the first
+            configuration in
+            [`new_receiver`][frequenz.sdk.config.ConfigManager.new_receiver] if
+            `wait_for_first` is `True`.
+
+    Returns:
+        The global instance of the ConfigManagingActor.
+
+    Raises:
+        RuntimeError: If the config manager is already initialized.
+    """
+    _logger.info(
+        "Initializing config manager %s for %s with events=%s, force_polling=%s, "
+        "polling_interval=%s, wait_for_first_timeout=%s",
+        name,
+        config_paths,
+        [event.name for event in event_types],
+        force_polling,
+        polling_interval,
+        wait_for_first_timeout,
+    )
+
+    global _CONFIG_MANAGER
+    if _CONFIG_MANAGER is not None:
+        raise RuntimeError("Config already initialized")
+
+    _CONFIG_MANAGER = ConfigManager(
+        config_paths,
+        event_types=event_types,
+        name=name,
+        force_polling=force_polling,
+        polling_interval=polling_interval,
+        wait_for_first_timeout=wait_for_first_timeout,
+        auto_start=True,
+    )
+
+    return _CONFIG_MANAGER
+
+
+async def shutdown_config_manager(
+    *,
+    msg: str = "Config manager is shutting down",
+    timeout: timedelta | None = timedelta(seconds=5),
+) -> None:
+    """Shutdown the global config manager.
+
+    This will stop the config manager and release any resources it holds.
+
+    Note:
+        The config manager must be
+        [initialized][frequenz.sdk.config.initialize_config] before calling this
+        function.
+
+    Args:
+        msg: The message to be passed to the tasks being cancelled.
+        timeout: The maximum time to wait for the config manager to stop. If `None`,
+            the method will only cancel the config manager actor and return immediately
+            without awaiting at all (stopping might continue in the background). If the
+            time is exceeded, an error will be logged.
+
+    Raises:
+        RuntimeError: If the config manager is not initialized.
+    """
+    _logger.info("Shutting down config manager (timeout=%s)...", timeout)
+
+    global _CONFIG_MANAGER
+    if _CONFIG_MANAGER is None:
+        raise RuntimeError("Config not initialized")
+
+    if timeout is None:
+        _CONFIG_MANAGER.actor.cancel(msg)
+        _logger.info(
+            "Config manager cancelled, stopping might continue in the background."
+        )
+    else:
+        try:
+            async with asyncio.timeout(timeout.total_seconds()):
+                await _CONFIG_MANAGER.actor.stop(msg)
+                _logger.info("Config manager stopped.")
+        except asyncio.TimeoutError:
+            _logger.warning(
+                "Config manager did not stop within %s seconds, it might continue "
+                "stopping in the background",
+                timeout,
+            )
+
+    _CONFIG_MANAGER = None
+
+
+def config_manager() -> ConfigManager:
+    """Return the global config manager.
+
+    Note:
+        The config manager must be
+        [initialized][frequenz.sdk.config.initialize_config] before calling this
+        function.
+
+    Returns:
+        The global instance of the config manager.
+
+    Raises:
+        RuntimeError: If the config manager is not initialized.
+    """
+    if _CONFIG_MANAGER is None:
+        raise RuntimeError("Config not initialized")
+    return _CONFIG_MANAGER