config

frequenz.sdk.config

Read and update config variables.

Classes

frequenz.sdk.config.ConfigManagingActor

Bases: Actor

An actor that monitors a TOML configuration file for updates.

When the file is updated, the new configuration is sent, as a dict, to the output sender.

When the actor is started, if a configuration file already exists, then it will be read and sent to the output sender before the actor starts monitoring the file for updates. This way users can rely on the actor to do the initial configuration reading too.

Source code in frequenz/sdk/config/_config_managing.py

class ConfigManagingActor(Actor):
    """An actor that monitors a TOML configuration file for updates.

    When the file is updated, the new configuration is sent, as a [`dict`][], to the
    `output` sender.

    When the actor is started, if a configuration file already exists, then it will be
    read and sent to the `output` sender before the actor starts monitoring the file
    for updates. This way users can rely on the actor to do the initial configuration
    reading too.
    """

    def __init__(
        self,
        config_path: pathlib.Path | str,
        output: Sender[abc.Mapping[str, Any]],
        event_types: abc.Set[EventType] = frozenset(EventType),
        *,
        name: str | None = None,
    ) -> None:
        """Initialize this instance.

        Args:
            config_path: The path to the TOML file with the configuration.
            output: The sender to send the configuration to.
            event_types: The set of event types to monitor.
            name: The name of the actor. If `None`, `str(id(self))` will
                be used. This is used mostly for debugging purposes.
        """
        super().__init__(name=name)
        self._config_path: pathlib.Path = (
            config_path
            if isinstance(config_path, pathlib.Path)
            else pathlib.Path(config_path)
        )
        self._output: Sender[abc.Mapping[str, Any]] = output
        self._event_types: abc.Set[EventType] = event_types

    def _read_config(self) -> abc.Mapping[str, Any]:
        """Read the contents of the configuration file.

        Returns:
            A dictionary containing configuration variables.

        Raises:
            ValueError: If config file cannot be read.
        """
        try:
            with self._config_path.open("rb") as toml_file:
                return tomllib.load(toml_file)
        except ValueError as err:
            _logger.error("%s: Can't read config file, err: %s", self, err)
            raise

    async def send_config(self) -> None:
        """Send the configuration to the output sender."""
        config = self._read_config()
        await self._output.send(config)

    async def _run(self) -> None:
        """Monitor for and send configuration file updates.

        At startup, the Config Manager sends the current config so that it
        can be cache in the Broadcast channel and served to receivers even if
        there hasn't been any change to the config file itself.
        """
        await self.send_config()

        # FileWatcher can't watch for non-existing files, so we need to watch for the
        # parent directory instead just in case a configuration file doesn't exist yet
        # or it is deleted and recreated again.
        file_watcher = FileWatcher(
            paths=[self._config_path.parent], event_types=self._event_types
        )

        try:
            async for event in file_watcher:
                # Since we are watching the whole parent directory, we need to make sure
                # we only react to events related to the configuration file.
                if not event.path.samefile(self._config_path):
                    continue

                match event.type:
                    case EventType.CREATE:
                        _logger.info(
                            "%s: The configuration file %s was created, sending new config...",
                            self,
                            self._config_path,
                        )
                        await self.send_config()
                    case EventType.MODIFY:
                        _logger.info(
                            "%s: The configuration file %s was modified, sending update...",
                            self,
                            self._config_path,
                        )
                        await self.send_config()
                    case EventType.DELETE:
                        _logger.info(
                            "%s: The configuration file %s was deleted, ignoring...",
                            self,
                            self._config_path,
                        )
                    case _:
                        assert_never(event.type)
        finally:
            del file_watcher

Attributes

RESTART_DELAY class-attribute instance-attribute

RESTART_DELAY: timedelta = timedelta(seconds=2)

The delay to wait between restarts of this actor.

is_running property

is_running: bool

Return whether this background service is running.

A service is considered running when at least one task is running.

RETURNS	DESCRIPTION
bool	Whether this background service is running.

name property

name: str

The name of this background service.

RETURNS	DESCRIPTION
str	The name of this background service.

tasks property

tasks: Set[Task[Any]]

Return the set of running tasks spawned by this background service.

Users typically should not modify the tasks in the returned set and only use them for informational purposes.

Danger

Changing the returned tasks may lead to unexpected behavior, don't do it unless the class explicitly documents it is safe to do so.

RETURNS	DESCRIPTION
Set[Task[Any]]	The set of running tasks spawned by this background service.

Functions

__aenter__ async

__aenter__() -> Self

Enter an async context.

Start this background service.

RETURNS	DESCRIPTION
Self	This background service.

Source code in frequenz/sdk/actor/_background_service.py

async def __aenter__(self) -> Self:
    """Enter an async context.

    Start this background service.

    Returns:
        This background service.
    """
    self.start()
    return self

__aexit__ async

__aexit__(
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None

Exit an async context.

Stop this background service.

PARAMETER	DESCRIPTION
exc_type	The type of the exception raised, if any. TYPE: type[BaseException] | None
exc_val	The exception raised, if any. TYPE: BaseException | None
exc_tb	The traceback of the exception raised, if any. TYPE: TracebackType | None

Source code in frequenz/sdk/actor/_background_service.py

async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit an async context.

    Stop this background service.

    Args:
        exc_type: The type of the exception raised, if any.
        exc_val: The exception raised, if any.
        exc_tb: The traceback of the exception raised, if any.
    """
    await self.stop()

__await__

__await__() -> Generator[None, None, None]

Await this background service.

An awaited background service will wait for all its tasks to finish.

RETURNS	DESCRIPTION
None	An implementation-specific generator for the awaitable.

Source code in frequenz/sdk/actor/_background_service.py

def __await__(self) -> collections.abc.Generator[None, None, None]:
    """Await this background service.

    An awaited background service will wait for all its tasks to finish.

    Returns:
        An implementation-specific generator for the awaitable.
    """
    return self.wait().__await__()

__del__

__del__() -> None

Destroy this instance.

Cancel all running tasks spawned by this background service.

Source code in frequenz/sdk/actor/_background_service.py

def __del__(self) -> None:
    """Destroy this instance.

    Cancel all running tasks spawned by this background service.
    """
    self.cancel("{self!r} was deleted")

__init__

__init__(
    config_path: Path | str,
    output: Sender[Mapping[str, Any]],
    event_types: Set[EventType] = frozenset(EventType),
    *,
    name: str | None = None
) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
config_path	The path to the TOML file with the configuration. TYPE: Path | str
output	The sender to send the configuration to. TYPE: Sender[Mapping[str, Any]]
event_types	The set of event types to monitor. TYPE: Set[EventType] DEFAULT: frozenset(EventType)
name	The name of the actor. If None, str(id(self)) will be used. This is used mostly for debugging purposes. TYPE: str | None DEFAULT: None

Source code in frequenz/sdk/config/_config_managing.py

def __init__(
    self,
    config_path: pathlib.Path | str,
    output: Sender[abc.Mapping[str, Any]],
    event_types: abc.Set[EventType] = frozenset(EventType),
    *,
    name: str | None = None,
) -> None:
    """Initialize this instance.

    Args:
        config_path: The path to the TOML file with the configuration.
        output: The sender to send the configuration to.
        event_types: The set of event types to monitor.
        name: The name of the actor. If `None`, `str(id(self))` will
            be used. This is used mostly for debugging purposes.
    """
    super().__init__(name=name)
    self._config_path: pathlib.Path = (
        config_path
        if isinstance(config_path, pathlib.Path)
        else pathlib.Path(config_path)
    )
    self._output: Sender[abc.Mapping[str, Any]] = output
    self._event_types: abc.Set[EventType] = event_types

__repr__

__repr__() -> str

Return a string representation of this instance.

RETURNS	DESCRIPTION
str	A string representation of this instance.

Source code in frequenz/sdk/actor/_background_service.py

def __repr__(self) -> str:
    """Return a string representation of this instance.

    Returns:
        A string representation of this instance.
    """
    return f"{type(self).__name__}(name={self._name!r}, tasks={self._tasks!r})"

__str__

__str__() -> str

Return a string representation of this instance.

RETURNS	DESCRIPTION
str	A string representation of this instance.

Source code in frequenz/sdk/actor/_background_service.py

def __str__(self) -> str:
    """Return a string representation of this instance.

    Returns:
        A string representation of this instance.
    """
    return f"{type(self).__name__}[{self._name}]"

cancel

cancel(msg: str | None = None) -> None

Cancel all running tasks spawned by this background service.

PARAMETER	DESCRIPTION
msg	The message to be passed to the tasks being cancelled. TYPE: str | None DEFAULT: None

Source code in frequenz/sdk/actor/_background_service.py

def cancel(self, msg: str | None = None) -> None:
    """Cancel all running tasks spawned by this background service.

    Args:
        msg: The message to be passed to the tasks being cancelled.
    """
    for task in self._tasks:
        task.cancel(msg)

send_config async

send_config() -> None

Send the configuration to the output sender.

Source code in frequenz/sdk/config/_config_managing.py

async def send_config(self) -> None:
    """Send the configuration to the output sender."""
    config = self._read_config()
    await self._output.send(config)

start

start() -> None

Start this actor.

If this actor is already running, this method does nothing.

Source code in frequenz/sdk/actor/_actor.py

def start(self) -> None:
    """Start this actor.

    If this actor is already running, this method does nothing.
    """
    if self.is_running:
        return
    self._tasks.clear()
    self._tasks.add(asyncio.create_task(self._run_loop()))

stop async

stop(msg: str | None = None) -> None

Stop this background service.

This method cancels all running tasks spawned by this service and waits for them to finish.

PARAMETER	DESCRIPTION
msg	The message to be passed to the tasks being cancelled. TYPE: str | None DEFAULT: None

RAISES	DESCRIPTION
BaseExceptionGroup	If any of the tasks spawned by this service raised an exception.

Source code in frequenz/sdk/actor/_background_service.py

async def stop(self, msg: str | None = None) -> None:
    """Stop this background service.

    This method cancels all running tasks spawned by this service and waits for them
    to finish.

    Args:
        msg: The message to be passed to the tasks being cancelled.

    Raises:
        BaseExceptionGroup: If any of the tasks spawned by this service raised an
            exception.
    """
    if not self._tasks:
        return
    self.cancel(msg)
    try:
        await self.wait()
    except BaseExceptionGroup as exc_group:
        # We want to ignore CancelledError here as we explicitly cancelled all the
        # tasks.
        _, rest = exc_group.split(asyncio.CancelledError)
        if rest is not None:
            # We are filtering out from an exception group, we really don't want to
            # add the exceptions we just filtered by adding a from clause here.
            raise rest  # pylint: disable=raise-missing-from

wait async

wait() -> None

Wait this background service to finish.

Wait until all background service tasks are finished.

RAISES	DESCRIPTION
BaseExceptionGroup	If any of the tasks spawned by this service raised an exception (CancelError is not considered an error and not returned in the exception group).

Source code in frequenz/sdk/actor/_background_service.py

async def wait(self) -> None:
    """Wait this background service to finish.

    Wait until all background service tasks are finished.

    Raises:
        BaseExceptionGroup: If any of the tasks spawned by this service raised an
            exception (`CancelError` is not considered an error and not returned in
            the exception group).
    """
    # We need to account for tasks that were created between when we started
    # awaiting and we finished awaiting.
    while self._tasks:
        done, pending = await asyncio.wait(self._tasks)
        assert not pending

        # We remove the done tasks, but there might be new ones created after we
        # started waiting.
        self._tasks = self._tasks - done

        exceptions: list[BaseException] = []
        for task in done:
            try:
                # This will raise a CancelledError if the task was cancelled or any
                # other exception if the task raised one.
                _ = task.result()
            except BaseException as error:  # pylint: disable=broad-except
                exceptions.append(error)
        if exceptions:
            raise BaseExceptionGroup(
                f"Error while stopping background service {self}", exceptions
            )