warnings

frequenz.core.warnings

Warnings utilities.

This module provides ignoring_warnings, a context manager to silence warnings around a piece of code, without the side effect that warnings.catch_warnings has, and ignoring_deprecations for the common case of a library having to touch a symbol it deprecated itself.

It also provides deprecated_aliases, to keep the old import path of a symbol that moved to another module, or was renamed, working, warning whoever uses it since which version it is deprecated, as its DeprecatedAlias entry says, and asserting_no_warnings with its asserting_no_deprecations shortcut, to check in a test that a piece of code doesn't warn.

The documented way of silencing a warning locally is a warnings.catch_warnings block, but merely entering and leaving one invalidates the warnings deduplication history of the whole program, so every warning that was already shown is shown again, and again on every call. This is python/cpython#73858, open since 2017.

ignoring_warnings adds the filter to the list the interpreter consults and takes it out again on exit, which leaves the deduplication history untouched, for about half a microsecond more per block than the standard library one costs:

import warnings

from frequenz.core.warnings import ignoring_warnings


def legacy_convert(value: str) -> int:
    warnings.warn("legacy_convert() is deprecated", DeprecationWarning, stacklevel=2)
    return int(value)


def convert(value: str) -> int:
    with ignoring_warnings(category=DeprecationWarning):
        return legacy_convert(value)


with warnings.catch_warnings(record=True, action="default") as caught:
    for _ in range(10):
        warnings.warn("shown only once", UserWarning)
        convert("1")

assert len(caught) == 1

With warnings.catch_warnings in convert() the same loop shows the UserWarning ten times.

Classes

frequenz.core.warnings.DeprecatedAlias dataclass

A symbol kept reachable from its old name.

Each one is an entry for deprecated_aliases, which has the examples.

The symbol moved to another module, new_module, was renamed, new_name, or both. Without new_module it is still in the module defining the alias, under its new name.

An alias says either since which version it is deprecated, with since, or its whole warning message, with message, but not both. since="v1.2.0" warns with <old> is deprecated since v1.2.0. Use <new> instead., which is what the deprecations guide suggests, so message is only needed to say something else.

Everything is checked when the alias is created rather than when it is reached.

Source code in src/frequenz/core/warnings/_deprecated_aliases.py

@dataclass(frozen=True, init=False, slots=True)
class DeprecatedAlias:
    """A symbol kept reachable from its old name.

    Each one is an entry for [`deprecated_aliases`][..deprecated_aliases], which
    has the examples.

    The symbol moved to another module, `new_module`, was renamed, `new_name`, or
    both. Without `new_module` it is still in the module defining the alias, under
    its new name.

    An alias says either since which version it is deprecated, with `since`, or
    its whole warning message, with `message`, but not both. `since="v1.2.0"`
    warns with `<old> is deprecated since v1.2.0. Use <new> instead.`, which is
    what the deprecations guide suggests, so `message` is only needed to say
    something else.

    Everything is checked when the alias is created rather than when it is
    reached.
    """

    name: str
    """The deprecated name, in the module defining the alias."""

    new_module: str | None
    """The fully qualified name of the module the symbol lives in now.

    `None` if it is still in the module defining the alias, under
    [`new_name`][..new_name].
    """

    new_name: str
    """The name the symbol has now, [`name`][..name] if it was not renamed."""

    since: str | None
    """The version the alias is deprecated since, or `None` if it has a `message`.

    The warning says `<old> is deprecated since <since>. Use <new> instead.`
    """

    message: str | None
    """The template for the warning message, or `None` if it has a `since`.

    It is formatted with `{old}` and `{new}`, the fully qualified names of the
    alias and of the symbol it resolves to, as plain fields, without attributes,
    indexes, conversions or format specs.
    """

    # One pair of overloads with `new_module` and one with only `new_name`, so a call
    # with neither matches none of them.
    @overload
    def __init__(  # noqa: D107
        self,
        name: str,
        /,
        *,
        new_module: str,
        new_name: str | None = None,
        since: str,
    ) -> None: ...

    @overload
    def __init__(  # noqa: D107
        self,
        name: str,
        /,
        *,
        new_module: str,
        new_name: str | None = None,
        message: str,
    ) -> None: ...

    @overload
    def __init__(  # noqa: D107
        self,
        name: str,
        /,
        *,
        new_module: None = None,
        new_name: str,
        since: str,
    ) -> None: ...

    @overload
    def __init__(  # noqa: D107
        self,
        name: str,
        /,
        *,
        new_module: None = None,
        new_name: str,
        message: str,
    ) -> None: ...

    def __init__(  # noqa: DOC502
        self,
        name: str,
        /,
        *,
        new_module: str | None = None,
        new_name: str | None = None,
        since: str | None = None,
        message: str | None = None,
    ) -> None:
        """Initialize this instance.

        At least one of `new_module` and `new_name` must be given, and exactly one
        of `since` and `message`.

        Args:
            name: The deprecated name, in the module defining the alias.
            new_module: The fully qualified name of the module the symbol lives in
                now, if it moved out of the module defining the alias.
            new_name: The name the symbol has now, if it was renamed.
            since: The version the alias is deprecated since, such as `"v1.2.0"`.
                The warning then says `<old> is deprecated since <since>. Use
                <new> instead.`
            message: The template for the whole warning message instead, formatted
                with `{old}` and `{new}`, the fully qualified names of the alias
                and of the symbol it resolves to, as plain fields, without
                attributes, indexes, conversions or format specs.

        Raises:
            TypeError: If an argument is not a string, if neither `new_module` nor
                `new_name` is given, or if not exactly one of `since` and
                `message` is.
            ValueError: If `name`, `new_name` or a part of `new_module` is not an
                identifier, if `since` is empty, or if `message` is not a template
                with only plain `{old}` and `{new}` fields.
        """
        # The class is frozen, which blocks plain assignment here too.
        object.__setattr__(self, "name", name)
        object.__setattr__(self, "new_module", new_module)
        object.__setattr__(self, "new_name", name if new_name is None else new_name)
        object.__setattr__(self, "since", since)
        object.__setattr__(self, "message", message)
        self._validate(new_name)

    def format_message(self, *, old: str, new: str) -> str:
        """Format the warning message for this alias.

        Args:
            old: The fully qualified name of the alias.
            new: The fully qualified name of the symbol it resolves to.

        Returns:
            `<old> is deprecated since <since>. Use <new> instead.` if the alias
                has a `since`, or its `message` formatted with `old` and `new`.
        """
        if self.message is None:
            return f"{old} is deprecated since {self.since}. Use {new} instead."
        return self.message.format(old=old, new=new)

    def _validate(self, new_name: str | None) -> None:
        """Check the fields.

        Args:
            new_name: The `new_name` given to `__init__()`, before it defaulted to
                `name`, to tell whether it was given at all.

        Raises:
            TypeError: If neither `new_module` nor `new_name` is given, or if not
                exactly one of `since` and `message` is.
        """
        if self.new_module is None and new_name is None:
            raise TypeError(
                f"the alias {self.name!r} needs new_module, new_name or both, "
                "got neither"
            )
        if (self.since is None) == (self.message is None):
            raise TypeError(
                f"the alias {self.name!r} needs exactly one of since and message, "
                f"got {'both' if self.since is not None else 'neither'}"
            )
        self._validate_name()
        self._validate_new_module()
        self._validate_new_name()
        self._validate_since()
        self._validate_message()

    def _validate_name(self) -> None:
        """Check that `name` is an identifier.

        Raises:
            TypeError: If it is not a string.
            ValueError: If it is not an identifier.
        """
        if not isinstance(self.name, str):
            raise TypeError(f"alias names must be str, got {self.name!r}")
        if not self.name.isidentifier():
            raise ValueError(f"alias names must be identifiers, got {self.name!r}")

    def _validate_new_module(self) -> None:
        """Check that `new_module`, if given, is a module name.

        Raises:
            TypeError: If it is not a string.
            ValueError: If one of its dotted parts is not an identifier.
        """
        if self.new_module is None:
            return
        if not isinstance(self.new_module, str):
            raise TypeError(
                f"the new_module of {self.name!r} must be a str, "
                f"got {self.new_module!r}"
            )
        if not all(part.isidentifier() for part in self.new_module.split(".")):
            raise ValueError(
                f"the new_module of {self.name!r} is not a module name: "
                f"{self.new_module!r}"
            )

    def _validate_new_name(self) -> None:
        """Check that `new_name` is an identifier.

        Raises:
            TypeError: If it is not a string.
            ValueError: If it is not an identifier.
        """
        if not isinstance(self.new_name, str):
            raise TypeError(
                f"the new_name of {self.name!r} must be a str, got {self.new_name!r}"
            )
        if not self.new_name.isidentifier():
            raise ValueError(
                f"the new_name of {self.name!r} is not an identifier: "
                f"{self.new_name!r}"
            )

    def _validate_since(self) -> None:
        """Check that `since`, if given, is not empty.

        Raises:
            TypeError: If it is not a string.
            ValueError: If it is empty or blank.
        """
        if self.since is None:
            return
        if not isinstance(self.since, str):
            raise TypeError(
                f"the since of {self.name!r} must be a str, got {self.since!r}"
            )
        if not self.since.strip():
            raise ValueError(f"the since of {self.name!r} is empty")

    def _validate_message(self) -> None:
        """Check that `message`, if given, is a template with plain `{old}` and `{new}`.

        Raises:
            TypeError: If it is not a string.
            ValueError: If it has a stray brace, or a replacement field other than a
                plain `{old}` or `{new}`.
        """
        if self.message is None:
            return
        if not isinstance(self.message, str):
            raise TypeError(
                f"the message of {self.name!r} must be a str, got {self.message!r}"
            )
        problem = (
            f"the message of {self.name!r} must be a template with only plain "
            f"{{old}} and {{new}} fields, got {self.message!r}"
        )
        try:
            # A stray brace raises here.
            parsed = list(string.Formatter().parse(self.message))
        except ValueError as error:
            raise ValueError(problem) from error
        # Anything more than a plain field can fail depending on the names it is
        # formatted with, which are only known when the alias is reached, so
        # formatting it once here can't tell: `{old.missing}` raises
        # `AttributeError`, and `{old:{new}}` makes the new name a format spec.
        if not all(
            field in (None, "old", "new") and not spec and conversion is None
            for _, field, spec, conversion in parsed
        ):
            raise ValueError(problem)

Attributes

message instance-attribute

message: str | None

The template for the warning message, or None if it has a since.

It is formatted with {old} and {new}, the fully qualified names of the alias and of the symbol it resolves to, as plain fields, without attributes, indexes, conversions or format specs.

name instance-attribute

name: str

The deprecated name, in the module defining the alias.

new_module instance-attribute

new_module: str | None

The fully qualified name of the module the symbol lives in now.

None if it is still in the module defining the alias, under new_name.

new_name instance-attribute

new_name: str

The name the symbol has now, name if it was not renamed.

since instance-attribute

since: str | None

The version the alias is deprecated since, or None if it has a message.

The warning says <old> is deprecated since <since>. Use <new> instead.

Methods:

__init__

__init__(
    name: str,
    /,
    *,
    new_module: str,
    new_name: str | None = None,
    since: str,
) -> None

__init__(
    name: str,
    /,
    *,
    new_module: str,
    new_name: str | None = None,
    message: str,
) -> None

__init__(
    name: str,
    /,
    *,
    new_module: None = None,
    new_name: str,
    since: str,
) -> None

__init__(
    name: str,
    /,
    *,
    new_module: None = None,
    new_name: str,
    message: str,
) -> None

__init__(
    name: str,
    /,
    *,
    new_module: str | None = None,
    new_name: str | None = None,
    since: str | None = None,
    message: str | None = None,
) -> None

Initialize this instance.

At least one of new_module and new_name must be given, and exactly one of since and message.

PARAMETER	DESCRIPTION
name	The deprecated name, in the module defining the alias. TYPE: str
new_module	The fully qualified name of the module the symbol lives in now, if it moved out of the module defining the alias. TYPE: str | None DEFAULT: None
new_name	The name the symbol has now, if it was renamed. TYPE: str | None DEFAULT: None
since	The version the alias is deprecated since, such as "v1.2.0". The warning then says <old> is deprecated since <since>. Use <new> instead. TYPE: str | None DEFAULT: None
message	The template for the whole warning message instead, formatted with {old} and {new}, the fully qualified names of the alias and of the symbol it resolves to, as plain fields, without attributes, indexes, conversions or format specs. TYPE: str | None DEFAULT: None

RAISES	DESCRIPTION
TypeError	If an argument is not a string, if neither new_module nor new_name is given, or if not exactly one of since and message is.
ValueError	If name, new_name or a part of new_module is not an identifier, if since is empty, or if message is not a template with only plain {old} and {new} fields.

Source code in src/frequenz/core/warnings/_deprecated_aliases.py

def __init__(  # noqa: DOC502
    self,
    name: str,
    /,
    *,
    new_module: str | None = None,
    new_name: str | None = None,
    since: str | None = None,
    message: str | None = None,
) -> None:
    """Initialize this instance.

    At least one of `new_module` and `new_name` must be given, and exactly one
    of `since` and `message`.

    Args:
        name: The deprecated name, in the module defining the alias.
        new_module: The fully qualified name of the module the symbol lives in
            now, if it moved out of the module defining the alias.
        new_name: The name the symbol has now, if it was renamed.
        since: The version the alias is deprecated since, such as `"v1.2.0"`.
            The warning then says `<old> is deprecated since <since>. Use
            <new> instead.`
        message: The template for the whole warning message instead, formatted
            with `{old}` and `{new}`, the fully qualified names of the alias
            and of the symbol it resolves to, as plain fields, without
            attributes, indexes, conversions or format specs.

    Raises:
        TypeError: If an argument is not a string, if neither `new_module` nor
            `new_name` is given, or if not exactly one of `since` and
            `message` is.
        ValueError: If `name`, `new_name` or a part of `new_module` is not an
            identifier, if `since` is empty, or if `message` is not a template
            with only plain `{old}` and `{new}` fields.
    """
    # The class is frozen, which blocks plain assignment here too.
    object.__setattr__(self, "name", name)
    object.__setattr__(self, "new_module", new_module)
    object.__setattr__(self, "new_name", name if new_name is None else new_name)
    object.__setattr__(self, "since", since)
    object.__setattr__(self, "message", message)
    self._validate(new_name)

format_message

format_message(*, old: str, new: str) -> str

Format the warning message for this alias.

PARAMETER	DESCRIPTION
old	The fully qualified name of the alias. TYPE: str
new	The fully qualified name of the symbol it resolves to. TYPE: str

RETURNS	DESCRIPTION
str	<old> is deprecated since <since>. Use <new> instead. if the alias has a since, or its message formatted with old and new.

Source code in src/frequenz/core/warnings/_deprecated_aliases.py

def format_message(self, *, old: str, new: str) -> str:
    """Format the warning message for this alias.

    Args:
        old: The fully qualified name of the alias.
        new: The fully qualified name of the symbol it resolves to.

    Returns:
        `<old> is deprecated since <since>. Use <new> instead.` if the alias
            has a `since`, or its `message` formatted with `old` and `new`.
    """
    if self.message is None:
        return f"{old} is deprecated since {self.since}. Use {new} instead."
    return self.message.format(old=old, new=new)

frequenz.core.warnings.asserting_no_warnings

Bases: AbstractContextManager[None]

A context manager that fails if a matching warning is raised inside its block.

Use this to test that a call doesn't warn. The usual alternative, an "error" filter, makes warnings.warn raise an exception inside the code under test. A broad except in that code can then swallow it or turn it into an unrelated failure, and the code no longer behaves as it would in production. This block records warnings instead and leaves the code alone. When the block ends, it raises AssertionError listing every unexpected warning and where it came from.

Warnings that don't cause a failure are still shown as usual when the block ends. The unexpected ones are shown too if the block is left because of an exception. In that case the exception is raised instead of the AssertionError, since it's the bigger problem, but the warnings aren't lost.

Resets the warnings deduplication history

Recording needs warnings.catch_warnings, and using it resets that history for the whole program. There's no way around it here: a warning that was already shown is skipped (by the module's __warningregistry__) before the filters are even checked, so the only way to catch it again is to reset the history.

In tests this is usually fine, since they are isolated. Anywhere else, every warning shown before the block will be shown again after it. So use this in a short test that checks just this, not in a hot path, a long-running process, or around a whole test suite.

Concurrency

warnings.catch_warnings affects the whole process unless sys.flags.context_aware_warnings is on. That flag was added in Python 3.14 and is off by default, except in free-threaded builds. Without it, a matching warning from another thread, or from another asyncio task while the block is waiting on an await, is recorded here and fails the block, even if the code under test never warned. Use it only around synchronous code, and not where background threads can raise matching warnings.

With the flag on, the problem is smaller but not gone. New threads start with an empty context, so their warnings aren't recorded. Tasks created inside the block, however, inherit its recording and can still fail it.

An ignoring_warnings block inside wins

ignoring_warnings puts its filter in front of the one this block adds. So if the code under test ignores a warning itself, that warning is never recorded and can't fail the block. This is intended, since the code explicitly asked to ignore it. It does mean that an asserting_no_deprecations block around code using ignoring_deprecations internally only checks the deprecations the code didn't silence, not all of them.

Handlers installed inside the block

While the block is active, it controls how warnings are shown. Handlers installed before the block, like the one logging.captureWarnings sets up, keep working: once the block ends, they receive the warnings that didn't fail it. But if the code under test replaces warnings.showwarning itself, any warning raised after that goes straight to its own handler, and this block can't see it or fail on it. A plain catch_warnings(record=True) has the same limitation.

Not reentrant

Entering an instance that is already inside its block raises RuntimeError.

Example

import warnings

from frequenz.core.warnings import asserting_no_warnings


def convert(value: str) -> int:
    return int(value)


with asserting_no_warnings(category=DeprecationWarning):
    assert convert("1") == 1

Source code in src/frequenz/core/warnings/_asserting.py

class asserting_no_warnings(contextlib.AbstractContextManager[None]):
    """A context manager that fails if a matching warning is raised inside its block.

    Use this to test that a call doesn't warn. The usual alternative, an
    `"error"` filter, makes [`warnings.warn`][] raise an exception **inside** the
    code under test. A broad `except` in that code can then swallow it or turn
    it into an unrelated failure, and the code no longer behaves as it would in
    production. This block records warnings instead and leaves the code alone.
    When the block ends, it raises [`AssertionError`][] listing every
    unexpected warning and where it came from.

    Warnings that don't cause a failure are still shown as usual when the block
    ends. The unexpected ones are shown too if the block is left because of an
    exception. In that case the exception is raised instead of the
    [`AssertionError`][], since it's the bigger problem, but the warnings aren't
    lost.

    Warning: Resets the warnings deduplication history
        Recording needs [`warnings.catch_warnings`][], and using it resets that
        history for the whole program. There's no way around it here: a warning
        that was already shown is skipped (by the module's
        `__warningregistry__`) before the filters are even checked, so the only
        way to catch it again is to reset the history.

        In tests this is usually fine, since they are isolated. Anywhere else,
        every warning shown before the block will be shown again after it. So
        use this in a short test that checks just this, not in a hot path, a
        long-running process, or around a whole test suite.

    Warning: Concurrency
        [`warnings.catch_warnings`][] affects the whole process unless
        `sys.flags.context_aware_warnings` is on. That flag was added in Python
        3.14 and is off by default, except in free-threaded builds. Without it,
        a matching warning from another thread, or from another asyncio task
        while the block is waiting on an `await`, is recorded here and fails the
        block, even if the code under test never warned. Use it only around
        synchronous code, and not where background threads can raise matching
        warnings.

        With the flag on, the problem is smaller but not gone. New threads start
        with an empty context, so their warnings aren't recorded. Tasks created
        inside the block, however, inherit its recording and can still fail it.

    Warning: An `ignoring_warnings` block inside wins
        [`ignoring_warnings`][..ignoring_warnings] puts its filter in front of
        the one this block adds. So if the code under test ignores a warning
        itself, that warning is never recorded and can't fail the block. This is
        intended, since the code explicitly asked to ignore it. It does mean
        that an [`asserting_no_deprecations`][..asserting_no_deprecations] block
        around code using [`ignoring_deprecations`][..ignoring_deprecations]
        internally only checks the deprecations the code didn't silence, not all
        of them.

    Warning: Handlers installed inside the block
        While the block is active, it controls how warnings are shown. Handlers
        installed *before* the block, like the one [`logging.captureWarnings`][]
        sets up, keep working: once the block ends, they receive the warnings
        that didn't fail it. But if the code **under test** replaces
        [`warnings.showwarning`][] itself, any warning raised after that goes
        straight to its own handler, and this block can't see it or fail on it.
        A plain `catch_warnings(record=True)` has the same limitation.

    Warning: Not reentrant
        Entering an instance that is already inside its block raises
        [`RuntimeError`][].

    Example:
        ```python
        import warnings

        from frequenz.core.warnings import asserting_no_warnings


        def convert(value: str) -> int:
            return int(value)


        with asserting_no_warnings(category=DeprecationWarning):
            assert convert("1") == 1
        ```
    """

    def __init__(  # noqa: DOC503
        self, *, category: type[Warning] = Warning, message: str = ""
    ) -> None:
        """Initialize this instance.

        Args:
            category: The category of warnings to reject, including its subclasses.
            message: A regular expression the start of the warning message must
                match, case insensitively. The default matches every message.

        Raises:
            TypeError: If any argument has the wrong type.
            re.error: If `message` is not a valid regular expression.
        """
        if not (isinstance(category, type) and issubclass(category, Warning)):
            raise TypeError(f"category must be a Warning subclass, got {category!r}")
        if not isinstance(message, str):
            raise TypeError(f"message must be a str, got {message!r}")
        self._category = category
        self._message = message
        self._pattern = re.compile(message, re.IGNORECASE) if message else None
        self._entered = False
        self._stack: contextlib.ExitStack | None = None
        self._caught: list[warnings.WarningMessage] = []

    def _matches(self, warning: warnings.WarningMessage) -> bool:
        """Tell whether a recorded warning is one this block rejects.

        Args:
            warning: The recorded warning.

        Returns:
            Whether it matches.
        """
        # There is no `module` argument next to `category` and `message`, even
        # though `ignoring_warnings` has one and the "always" filter added in
        # `__enter__()` would take it, and it can't be added here as it stands.
        # A recorded `WarningMessage` carries `filename` and not the module
        # name the filter matched, and those are different strings (`mymod`
        # against `/tmp/mymod.py`), so this function can't reproduce the
        # decision. Putting it only on the filter doesn't work either: a
        # warning from a module the argument was meant to exempt still reaches
        # this function through an ambient "default" or "always" filter, and
        # would be reported. Giving it filename semantics instead would mean
        # one name for two meanings across this module.
        if not issubclass(warning.category, self._category):
            return False
        return self._pattern is None or bool(self._pattern.match(str(warning.message)))

    def __enter__(self) -> None:
        """Enter the block, recording the warnings raised inside it.

        Raises:
            RuntimeError: If this instance is already inside its block.
        """
        if self._entered:
            raise RuntimeError(f"Cannot enter {type(self).__name__}() twice")
        self._entered = True

        self._stack = contextlib.ExitStack()
        self._caught = self._stack.enter_context(warnings.catch_warnings(record=True))

        # Recording happens in the default warnings output, so it only sees a
        # warning while that output is the one installed. `catch_warnings` resets
        # `showwarning` itself to make sure of that, but only when the filters are
        # process-wide: with context-aware warnings (Python 3.14 and later) the
        # recording lives in the context while `showwarning` stays a module global,
        # so a handler the application installed, as `logging.captureWarnings()`
        # does, keeps taking every warning, nothing is recorded, and every assertion
        # here passes. That is python/cpython#151149, so the reset is done here too,
        # in every mode. It is undone before the recording block closes, so the
        # replay on the way out still reaches the application's handler.
        default_show = getattr(warnings, "_showwarning_orig", None)
        if default_show is not None:
            self._stack.callback(setattr, warnings, "showwarning", warnings.showwarning)
            warnings.showwarning = default_show

        # "always" so a warning already shown elsewhere is still recorded here, with
        # the message too, so a warning this block doesn't reject keeps whatever the
        # surrounding filters say about it.
        warnings.filterwarnings(
            "always", message=self._message, category=self._category
        )

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        """Exit the block, replaying what it didn't reject and failing on the rest.

        Args:
            exc_type: The type of the exception leaving the block, if any.
            exc_value: The exception leaving the block, if any.
            traceback: The traceback of that exception, if any.

        Raises:
            AssertionError: If a matching warning was raised inside the block and no
                exception is on its way out, which is the more important news.
        """
        self._entered = False
        stack, self._stack = self._stack, None
        caught, self._caught = self._caught, []
        if stack is not None:
            stack.close()

        # An exception is the more important news, so the assertion stands down and
        # the warnings it would have reported are shown instead of being reported.
        # Dropping them here would lose them entirely, which is the one thing the
        # replay below exists to prevent.
        failing = exc_type is None

        # One pass, rather than a list of offenders and a membership test per
        # warning: the filter added inside is an "always" one, so a regression in a
        # loop can easily leave tens of thousands of warnings here, and the failure
        # path is the last place to spend quadratic time.
        unexpected: list[warnings.WarningMessage] = []
        for warning in caught:
            if failing and self._matches(warning):
                unexpected.append(warning)
            else:
                _replay(warning)

        if unexpected:
            listing = "\n".join(
                f"  {warning.category.__name__}: {warning.message} "
                f"({warning.filename}:{warning.lineno})"
                for warning in unexpected
            )
            raise AssertionError(f"Unexpected warnings raised:\n{listing}")

Methods:

__enter__

__enter__() -> None

Enter the block, recording the warnings raised inside it.

RAISES	DESCRIPTION
RuntimeError	If this instance is already inside its block.

Source code in src/frequenz/core/warnings/_asserting.py

def __enter__(self) -> None:
    """Enter the block, recording the warnings raised inside it.

    Raises:
        RuntimeError: If this instance is already inside its block.
    """
    if self._entered:
        raise RuntimeError(f"Cannot enter {type(self).__name__}() twice")
    self._entered = True

    self._stack = contextlib.ExitStack()
    self._caught = self._stack.enter_context(warnings.catch_warnings(record=True))

    # Recording happens in the default warnings output, so it only sees a
    # warning while that output is the one installed. `catch_warnings` resets
    # `showwarning` itself to make sure of that, but only when the filters are
    # process-wide: with context-aware warnings (Python 3.14 and later) the
    # recording lives in the context while `showwarning` stays a module global,
    # so a handler the application installed, as `logging.captureWarnings()`
    # does, keeps taking every warning, nothing is recorded, and every assertion
    # here passes. That is python/cpython#151149, so the reset is done here too,
    # in every mode. It is undone before the recording block closes, so the
    # replay on the way out still reaches the application's handler.
    default_show = getattr(warnings, "_showwarning_orig", None)
    if default_show is not None:
        self._stack.callback(setattr, warnings, "showwarning", warnings.showwarning)
        warnings.showwarning = default_show

    # "always" so a warning already shown elsewhere is still recorded here, with
    # the message too, so a warning this block doesn't reject keeps whatever the
    # surrounding filters say about it.
    warnings.filterwarnings(
        "always", message=self._message, category=self._category
    )

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None

Exit the block, replaying what it didn't reject and failing on the rest.

PARAMETER	DESCRIPTION
exc_type	The type of the exception leaving the block, if any. TYPE: type[BaseException] | None
exc_value	The exception leaving the block, if any. TYPE: BaseException | None
traceback	The traceback of that exception, if any. TYPE: TracebackType | None

RAISES	DESCRIPTION
AssertionError	If a matching warning was raised inside the block and no exception is on its way out, which is the more important news.

Source code in src/frequenz/core/warnings/_asserting.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    """Exit the block, replaying what it didn't reject and failing on the rest.

    Args:
        exc_type: The type of the exception leaving the block, if any.
        exc_value: The exception leaving the block, if any.
        traceback: The traceback of that exception, if any.

    Raises:
        AssertionError: If a matching warning was raised inside the block and no
            exception is on its way out, which is the more important news.
    """
    self._entered = False
    stack, self._stack = self._stack, None
    caught, self._caught = self._caught, []
    if stack is not None:
        stack.close()

    # An exception is the more important news, so the assertion stands down and
    # the warnings it would have reported are shown instead of being reported.
    # Dropping them here would lose them entirely, which is the one thing the
    # replay below exists to prevent.
    failing = exc_type is None

    # One pass, rather than a list of offenders and a membership test per
    # warning: the filter added inside is an "always" one, so a regression in a
    # loop can easily leave tens of thousands of warnings here, and the failure
    # path is the last place to spend quadratic time.
    unexpected: list[warnings.WarningMessage] = []
    for warning in caught:
        if failing and self._matches(warning):
            unexpected.append(warning)
        else:
            _replay(warning)

    if unexpected:
        listing = "\n".join(
            f"  {warning.category.__name__}: {warning.message} "
            f"({warning.filename}:{warning.lineno})"
            for warning in unexpected
        )
        raise AssertionError(f"Unexpected warnings raised:\n{listing}")

__init__

__init__(
    *, category: type[Warning] = Warning, message: str = ""
) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
category	The category of warnings to reject, including its subclasses. TYPE: type[Warning] DEFAULT: Warning
message	A regular expression the start of the warning message must match, case insensitively. The default matches every message. TYPE: str DEFAULT: ''

RAISES	DESCRIPTION
TypeError	If any argument has the wrong type.
error	If message is not a valid regular expression.

Source code in src/frequenz/core/warnings/_asserting.py

def __init__(  # noqa: DOC503
    self, *, category: type[Warning] = Warning, message: str = ""
) -> None:
    """Initialize this instance.

    Args:
        category: The category of warnings to reject, including its subclasses.
        message: A regular expression the start of the warning message must
            match, case insensitively. The default matches every message.

    Raises:
        TypeError: If any argument has the wrong type.
        re.error: If `message` is not a valid regular expression.
    """
    if not (isinstance(category, type) and issubclass(category, Warning)):
        raise TypeError(f"category must be a Warning subclass, got {category!r}")
    if not isinstance(message, str):
        raise TypeError(f"message must be a str, got {message!r}")
    self._category = category
    self._message = message
    self._pattern = re.compile(message, re.IGNORECASE) if message else None
    self._entered = False
    self._stack: contextlib.ExitStack | None = None
    self._caught: list[warnings.WarningMessage] = []

frequenz.core.warnings.ignoring_warnings

Bases: AbstractContextManager[None]

A context manager that ignores the matching warnings raised inside its block.

The constructor arguments select which warnings are ignored, and mean the same as the arguments of warnings.filterwarnings with action="ignore".

Unlike warnings.catch_warnings, this doesn't reset the warnings deduplication history of the program, so warnings shown before the block are not shown again after it. See the module documentation for the background.

Example

import warnings

from frequenz.core.warnings import ignoring_warnings


# Some third-party function warns about something out of our control.
def third_party() -> int:
    warnings.warn("this is fine, actually", RuntimeWarning, stacklevel=2)
    return 42


def use_third_party() -> int:
    with ignoring_warnings(RuntimeWarning, message="this is fine"):
        return third_party()

This is not a catch_warnings replacement

This only adds a filter for the duration of the block, it doesn't save and restore the filters around it. Any change made to the filters inside the block, by this thread or another one, is still in effect afterwards, exactly as if the block wasn't there. Use warnings.catch_warnings when the point is to undo filter changes.

Scope

The filter is added to the filters the interpreter consults, which are normally shared by every thread and every asyncio task, so a matching warning raised by unrelated code running concurrently is ignored too, for as long as the block lasts. Keep the block around the call that needs it, and prefer not to hold it across an await.

Wrapping this in a warnings.catch_warnings block only narrows that on Python 3.14 and newer, with sys.flags.context_aware_warnings on, which is off by default except in free-threaded builds. The outer block then gives the current context a filters list of its own, kept in a contextvars.ContextVar, so other threads, and tasks created before it, are no longer affected. Tasks created inside it still inherit that same list, so they are silenced too. Without the flag, warnings.catch_warnings reaches for the same shared state and isolates nothing. Either way it resets the deduplication history, which is what this class exists to avoid.

Not reentrant

Entering an instance that is already inside its block raises RuntimeError. A with ignoring_warnings(...) statement builds a new instance each time, so this only comes up when one instance is kept in a variable and entered again from inside itself, by a recursive function or a helper called in the block. Nesting is rejected rather than supported because the inner block would only ignore what the outer one already ignores, so there is nothing to gain from making it work; an instance can be entered again once it has been left.

Not thread-safe

Taking the filter out again is a lookup followed by a deletion, with no locking in between, so another thread removing a filter from the same list at the same time can make this delete the wrong one, dropping a filter that was not this block's and leaving this block's behind. Nothing guards against that, because a lock would cost every block to protect against something only a program that changes warning filters from several threads at once can do. If yours does, don't use this.

Source code in src/frequenz/core/warnings/_ignoring.py

class ignoring_warnings(contextlib.AbstractContextManager[None]):
    """A context manager that ignores the matching warnings raised inside its block.

    The constructor arguments select which warnings are ignored, and mean the same as
    the arguments of [`warnings.filterwarnings`][] with `action="ignore"`.

    Unlike [`warnings.catch_warnings`][], this doesn't reset the warnings
    deduplication history of the program, so warnings shown before the block are not
    shown again after it. See the module documentation for the background.

    Example:
        ```python
        import warnings

        from frequenz.core.warnings import ignoring_warnings


        # Some third-party function warns about something out of our control.
        def third_party() -> int:
            warnings.warn("this is fine, actually", RuntimeWarning, stacklevel=2)
            return 42


        def use_third_party() -> int:
            with ignoring_warnings(RuntimeWarning, message="this is fine"):
                return third_party()
        ```

    Warning: This is not a `catch_warnings` replacement
        This only adds a filter for the duration of the block, it doesn't save and
        restore the filters around it. Any change made to the filters inside the
        block, by this thread or another one, is still in effect afterwards, exactly
        as if the block wasn't there. Use [`warnings.catch_warnings`][] when the
        point is to undo filter changes.

    Warning: Scope
        The filter is added to the filters the interpreter consults, which are
        normally shared by every thread and every asyncio task, so a matching
        warning raised by unrelated code running concurrently is ignored too, for as
        long as the block lasts. Keep the block around the call that needs it, and
        prefer not to hold it across an `await`.

        Wrapping this in a [`warnings.catch_warnings`][] block only narrows that on
        Python 3.14 and newer, with `sys.flags.context_aware_warnings` on, which is
        off by default except in free-threaded builds. The outer block then gives
        the current context a filters list of its own, kept in a
        [`contextvars.ContextVar`][], so other threads, and tasks created before
        it, are no longer affected. Tasks created inside it still inherit that same
        list, so they are silenced too. Without the flag,
        [`warnings.catch_warnings`][] reaches for the same shared state and
        isolates nothing. Either way it resets the deduplication history, which is
        what this class exists to avoid.

    Warning: Not reentrant
        Entering an instance that is already inside its block raises
        [`RuntimeError`][]. A `with ignoring_warnings(...)` statement builds a new
        instance each time, so this only comes up when one instance is kept in a
        variable and entered again from inside itself, by a recursive function or a
        helper called in the block. Nesting is rejected rather than supported
        because the inner block would only ignore what the outer one already
        ignores, so there is nothing to gain from making it work; an instance can
        be entered again once it has been left.

    Warning: Not thread-safe
        Taking the filter out again is a lookup followed by a deletion, with no
        locking in between, so another thread removing a filter from the same list
        at the same time can make this delete the wrong one, dropping a filter that
        was not this block's and leaving this block's behind. Nothing guards against
        that, because a lock would cost every block to protect against something
        only a program that changes warning filters from several threads at once can
        do. If yours does, don't use this.
    """

    def __init__(  # noqa: DOC502
        self,
        category: type[Warning] = Warning,
        *,
        message: str = "",
        module: str = "",
    ) -> None:
        """Initialize this instance.

        Args:
            category: The category of warnings to ignore, including its subclasses.
            message: A regular expression the start of the warning message must
                match, case insensitively. The default matches every message.
            module: A regular expression the start of the module name must match. The
                default matches every module.

        Raises:
            TypeError: If any argument has the wrong type.
            re.error: If `message` or `module` is not a valid regular expression.
        """
        self._arguments = (category, message, module)
        self._item = _build_filter(category, message, module)
        self._entered = False
        self._filters: list[Any] | None = None
        self._fallback: contextlib.ExitStack | None = None

    def __enter__(self) -> None:
        """Enter the block, adding the filter.

        Raises:
            RuntimeError: If this instance is already inside its block.
        """
        if self._entered:
            raise RuntimeError(f"Cannot enter {type(self).__name__}() twice")
        self._entered = True

        filters = _live_filters()

        if filters is None:
            # The interpreter doesn't keep its filters where we can reach them, so
            # there is nothing to be clever about: do what the standard library
            # does, which is correct, only more expensive and with the history
            # damage.
            category, message, module = self._arguments
            self._fallback = contextlib.ExitStack()
            self._fallback.enter_context(warnings.catch_warnings())
            warnings.filterwarnings(
                "ignore", message=message, category=category, module=module
            )
            return

        # The whole point: inserting into the live list directly means the internal
        # filters version doesn't move, and it is that version moving which makes
        # CPython treat every module's `__warningregistry__` as stale
        # (cpython#73858). Adding an "ignore" filter without bumping it is sound, and
        # only for "ignore": the registry is consulted *before* the filters, and an
        # ignore can only take warnings away, never add them, so a warning recorded
        # as already shown outside the block is still correctly recorded as shown
        # inside it, and the other way around. `simplefilter()` and
        # `filterwarnings()` can't do this, they bump the version on purpose because
        # they can add warnings back.
        #
        # The list object is remembered, rather than only looked up again on exit,
        # because code inside the block can replace `warnings.filters` with another
        # list and the entry has to come out of both (see `__exit__()`). Only our own
        # entry is removed, and nothing else is restored: an entry/exit snapshot
        # would make two blocks overlapping in different threads (A enters, B enters,
        # A exits, B exits) put each other's filters back, leaving one installed
        # forever.
        self._filters = filters
        filters.insert(0, self._item)

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_value: BaseException | None,
        traceback: TracebackType | None,
    ) -> None:
        """Exit the block, taking the filter out again.

        Args:
            exc_type: The type of the exception leaving the block, if any.
            exc_value: The exception leaving the block, if any.
            traceback: The traceback of that exception, if any.
        """
        self._entered = False
        fallback, self._fallback = self._fallback, None
        if fallback is not None:
            fallback.close()
            return
        filters, self._filters = self._filters, None
        if filters is None:
            return

        _remove_filter(filters, self._item)
        # Code inside the block may have replaced the filters in effect with a copy of
        # the list, which carries our entry along with the rest, and it is that copy
        # the interpreter consults now. Both lists are cleaned then: the live one
        # because leaving an ignore filter in it silences warnings after the block,
        # and the one entered with because it can be put back later, by the
        # `catch_warnings` block that made the copy. A list that is neither, left
        # behind by a second replacement inside the block, keeps our entry, and there
        # is no way to reach it to take it out.
        live = _live_filters()
        if live is not None and live is not filters:
            _remove_filter(live, self._item)

Methods:

__enter__

__enter__() -> None

Enter the block, adding the filter.

RAISES	DESCRIPTION
RuntimeError	If this instance is already inside its block.

Source code in src/frequenz/core/warnings/_ignoring.py

def __enter__(self) -> None:
    """Enter the block, adding the filter.

    Raises:
        RuntimeError: If this instance is already inside its block.
    """
    if self._entered:
        raise RuntimeError(f"Cannot enter {type(self).__name__}() twice")
    self._entered = True

    filters = _live_filters()

    if filters is None:
        # The interpreter doesn't keep its filters where we can reach them, so
        # there is nothing to be clever about: do what the standard library
        # does, which is correct, only more expensive and with the history
        # damage.
        category, message, module = self._arguments
        self._fallback = contextlib.ExitStack()
        self._fallback.enter_context(warnings.catch_warnings())
        warnings.filterwarnings(
            "ignore", message=message, category=category, module=module
        )
        return

    # The whole point: inserting into the live list directly means the internal
    # filters version doesn't move, and it is that version moving which makes
    # CPython treat every module's `__warningregistry__` as stale
    # (cpython#73858). Adding an "ignore" filter without bumping it is sound, and
    # only for "ignore": the registry is consulted *before* the filters, and an
    # ignore can only take warnings away, never add them, so a warning recorded
    # as already shown outside the block is still correctly recorded as shown
    # inside it, and the other way around. `simplefilter()` and
    # `filterwarnings()` can't do this, they bump the version on purpose because
    # they can add warnings back.
    #
    # The list object is remembered, rather than only looked up again on exit,
    # because code inside the block can replace `warnings.filters` with another
    # list and the entry has to come out of both (see `__exit__()`). Only our own
    # entry is removed, and nothing else is restored: an entry/exit snapshot
    # would make two blocks overlapping in different threads (A enters, B enters,
    # A exits, B exits) put each other's filters back, leaving one installed
    # forever.
    self._filters = filters
    filters.insert(0, self._item)

__exit__

__exit__(
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None

Exit the block, taking the filter out again.

PARAMETER	DESCRIPTION
exc_type	The type of the exception leaving the block, if any. TYPE: type[BaseException] | None
exc_value	The exception leaving the block, if any. TYPE: BaseException | None
traceback	The traceback of that exception, if any. TYPE: TracebackType | None

Source code in src/frequenz/core/warnings/_ignoring.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_value: BaseException | None,
    traceback: TracebackType | None,
) -> None:
    """Exit the block, taking the filter out again.

    Args:
        exc_type: The type of the exception leaving the block, if any.
        exc_value: The exception leaving the block, if any.
        traceback: The traceback of that exception, if any.
    """
    self._entered = False
    fallback, self._fallback = self._fallback, None
    if fallback is not None:
        fallback.close()
        return
    filters, self._filters = self._filters, None
    if filters is None:
        return

    _remove_filter(filters, self._item)
    # Code inside the block may have replaced the filters in effect with a copy of
    # the list, which carries our entry along with the rest, and it is that copy
    # the interpreter consults now. Both lists are cleaned then: the live one
    # because leaving an ignore filter in it silences warnings after the block,
    # and the one entered with because it can be put back later, by the
    # `catch_warnings` block that made the copy. A list that is neither, left
    # behind by a second replacement inside the block, keeps our entry, and there
    # is no way to reach it to take it out.
    live = _live_filters()
    if live is not None and live is not filters:
        _remove_filter(live, self._item)

__init__

__init__(
    category: type[Warning] = Warning,
    *,
    message: str = "",
    module: str = ""
) -> None

Initialize this instance.

PARAMETER	DESCRIPTION
category	The category of warnings to ignore, including its subclasses. TYPE: type[Warning] DEFAULT: Warning
message	A regular expression the start of the warning message must match, case insensitively. The default matches every message. TYPE: str DEFAULT: ''
module	A regular expression the start of the module name must match. The default matches every module. TYPE: str DEFAULT: ''

RAISES	DESCRIPTION
TypeError	If any argument has the wrong type.
error	If message or module is not a valid regular expression.

Source code in src/frequenz/core/warnings/_ignoring.py

def __init__(  # noqa: DOC502
    self,
    category: type[Warning] = Warning,
    *,
    message: str = "",
    module: str = "",
) -> None:
    """Initialize this instance.

    Args:
        category: The category of warnings to ignore, including its subclasses.
        message: A regular expression the start of the warning message must
            match, case insensitively. The default matches every message.
        module: A regular expression the start of the module name must match. The
            default matches every module.

    Raises:
        TypeError: If any argument has the wrong type.
        re.error: If `message` or `module` is not a valid regular expression.
    """
    self._arguments = (category, message, module)
    self._item = _build_filter(category, message, module)
    self._entered = False
    self._filters: list[Any] | None = None
    self._fallback: contextlib.ExitStack | None = None

Functions:

frequenz.core.warnings.asserting_no_deprecations

asserting_no_deprecations(
    *, message: str = ""
) -> asserting_no_warnings

Fail if a deprecation warning is raised inside the block.

This is asserting_no_warnings for the usual case, checking that the replacement for a deprecated symbol doesn't itself go through the deprecated one.

All limitations of asserting_no_warnings() apply

Read the documentation of asserting_no_warnings, many limitations apply here too, as this is only a thin wrapper around it.

Pass message when the code under test legitimately deprecates something else, or reaches a third-party deprecation there is nothing to be done about: the block wraps the code being tested, so unlike an ignoring_deprecations block it can't be narrowed by making it shorter.

Example

import warnings

from frequenz.core.warnings import asserting_no_deprecations


def convert(value: str) -> int:
    return int(value)


with asserting_no_deprecations(message="Wrapper is deprecated"):
    assert convert("1") == 1

PARAMETER	DESCRIPTION
message	A regular expression the start of the warning message must match, case insensitively. The default rejects every deprecation. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
asserting_no_warnings	A context manager that fails if a deprecation warning is raised in it.

RAISES	DESCRIPTION
TypeError	If message has the wrong type.
error	If message is not a valid regular expression.

Source code in src/frequenz/core/warnings/_asserting.py

def asserting_no_deprecations(  # noqa: DOC502
    *, message: str = ""
) -> asserting_no_warnings:
    """Fail if a deprecation warning is raised inside the block.

    This is [`asserting_no_warnings`][..asserting_no_warnings] for the usual case,
    checking that the replacement for a deprecated symbol doesn't itself go through
    the deprecated one.

    Warning: All limitations of `asserting_no_warnings()` apply
        Read the documentation of [`asserting_no_warnings`][..asserting_no_warnings],
        many limitations apply here too, as this is only a thin wrapper around it.

    Pass `message` when the code under test legitimately deprecates something else,
    or reaches a third-party deprecation there is nothing to be done about: the block
    wraps the code being tested, so unlike an
    [`ignoring_deprecations`][..ignoring_deprecations] block it can't be narrowed by
    making it shorter.

    Example:
        ```python
        import warnings

        from frequenz.core.warnings import asserting_no_deprecations


        def convert(value: str) -> int:
            return int(value)


        with asserting_no_deprecations(message="Wrapper is deprecated"):
            assert convert("1") == 1
        ```

    Args:
        message: A regular expression the start of the warning message must match,
            case insensitively. The default rejects every deprecation.

    Returns:
        A context manager that fails if a deprecation warning is raised in it.

    Raises:
        TypeError: If `message` has the wrong type.
        re.error: If `message` is not a valid regular expression.
    """
    return asserting_no_warnings(category=DeprecationWarning, message=message)

frequenz.core.warnings.deprecated_aliases

deprecated_aliases(
    module: str,
    /,
    *aliases: DeprecatedAlias,
    category: type[Warning] = DeprecationWarning,
    stacklevel: int = 2,
) -> Callable[[str], Any]

Build a module __getattr__ that warns about deprecated aliases.

Use this for symbols that moved to another module, or were renamed, but should keep working from their old import path, when a typing_extensions.deprecated decorator is not an option because the object is not yours to mark: decorating it would deprecate it for everybody, including the users of its new home or name. The alias also stays the very same object, so isinstance keeps working through both paths.

Each DeprecatedAlias says when that alias was deprecated or provides its own message, so aliases deprecated in different releases can emit different warnings.

Danger

Follow the usage example structure strictly. In particular never drop the else:, otherwise a type checker will see the __getattr__ assignment and treat every symbol in the module as Any, effectively disabling type checking for the entire module.

Usage

This is how a module that used to define Decimal itself, and now gets it from decimal, keeps the old import path working:

from typing import TYPE_CHECKING, TypeAlias

from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

if TYPE_CHECKING:
    # Private import, only for type checkers
    from decimal import Decimal as _Decimal

    Decimal: TypeAlias = _Decimal
else:
    __getattr__ = deprecated_aliases(
        __name__,
        DeprecatedAlias("Decimal", new_module="decimal", since="v1.2.0"),
    )

Reaching Decimal through this module, say mypkg.Decimal, now emits a DeprecationWarning saying mypkg.Decimal is deprecated since v1.2.0. Use decimal.Decimal instead.

Renaming

When the symbol was also renamed on the way out, new_name gives the name it has now:

from typing import TYPE_CHECKING, TypeAlias

from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

if TYPE_CHECKING:
    from fractions import Fraction as _Fraction

    Rational: TypeAlias = _Fraction
else:
    __getattr__ = deprecated_aliases(
        __name__,
        DeprecatedAlias(
            "Rational",
            new_module="fractions",
            new_name="Fraction",
            since="v1.3.0",
        ),
    )

Without new_module, the symbol was renamed in this very module:

from typing import TYPE_CHECKING, TypeAlias

from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases


class Widget:
    ...  # Called Gadget before v1.4.0


if TYPE_CHECKING:
    Gadget: TypeAlias = Widget
else:
    __getattr__ = deprecated_aliases(
        __name__,
        DeprecatedAlias("Gadget", new_name="Widget", since="v1.4.0"),
    )

Custom message

When the standard wording is not enough, an alias can give its whole message instead of since, as a template that gets the fully qualified old and new names as {old} and {new}:

from typing import TYPE_CHECKING, TypeAlias

from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

if TYPE_CHECKING:
    from decimal import Decimal as _Decimal

    Decimal: TypeAlias = _Decimal
else:
    __getattr__ = deprecated_aliases(
        __name__,
        DeprecatedAlias(
            "Decimal",
            new_module="decimal",
            message="{old} is deprecated since v1.2.0 and will be removed "
            "in v2.0.0. Use {new} instead.",
        ),
    )

Deprecating whole modules

This function can't be used to deprecate a whole module. If you need to do that, you can keep a real __init__.py at the old path, with a __getattr__ = deprecated_aliases(...) that includes all the symbols that used to be reachable there.

Security Warning

This function imports the modules the aliases point at, so the aliases are as trusted as an import statement in this module. Write them out as literals; don't build them from anything that comes from outside the program.

PARAMETER	DESCRIPTION
module	The fully qualified name of the module defining the aliases. TYPE: str
*aliases	The deprecated aliases, at least one, each with its own name. TYPE: DeprecatedAlias DEFAULT: ()
category	The category of the warning to emit. TYPE: type[Warning] DEFAULT: DeprecationWarning
stacklevel	How far up the stack the warning is reported, counting from the __getattr__ itself. The default of 2 points at the code reaching for the alias, and only needs raising if something wraps the returned function. TYPE: int DEFAULT: 2

RETURNS	DESCRIPTION
Callable[[str], Any]	A function suitable for use as the module's __getattr__.

RAISES	DESCRIPTION
TypeError	If an argument has the wrong type, including an alias that is not a DeprecatedAlias. Also later, when an alias is reached, if it resolves to a module rather than to a symbol in one.
ValueError	If there are no aliases, if two of them have the same name, if one points at another alias in the same module, itself included, or if stacklevel is smaller than 1.

Source code in src/frequenz/core/warnings/_deprecated_aliases.py

def deprecated_aliases(  # noqa: DOC502
    module: str,
    /,
    *aliases: DeprecatedAlias,
    category: type[Warning] = DeprecationWarning,
    stacklevel: int = 2,
) -> Callable[[str], Any]:
    """Build a module `__getattr__` that warns about deprecated aliases.

    Use this for symbols that moved to another module, or were renamed, but should
    keep working from their old import path, when a [`typing_extensions.deprecated`][]
    decorator is not an option because the object is not yours to mark: decorating
    it would deprecate it for everybody, including the users of its new home or name.
    The alias also stays the very same object, so [`isinstance`][] keeps working
    through both paths.

    Each [`DeprecatedAlias`][..DeprecatedAlias] says when that alias was deprecated
    or provides its own message, so aliases deprecated in different releases can
    emit different warnings.

    Danger:
        Follow the usage example structure strictly. In particular never drop
        the `else:`, otherwise a type checker will see the `__getattr__`
        assignment and treat every symbol in the module as [`Any`][typing.Any],
        effectively disabling type checking for the entire module.

    Example: Usage
        This is how a module that used to define `Decimal` itself, and now gets it
        from [`decimal`][], keeps the old import path working:

        ```python
        from typing import TYPE_CHECKING, TypeAlias

        from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

        if TYPE_CHECKING:
            # Private import, only for type checkers
            from decimal import Decimal as _Decimal

            Decimal: TypeAlias = _Decimal
        else:
            __getattr__ = deprecated_aliases(
                __name__,
                DeprecatedAlias("Decimal", new_module="decimal", since="v1.2.0"),
            )
        ```

        Reaching `Decimal` through this module, say `mypkg.Decimal`, now emits a
        `DeprecationWarning` saying `mypkg.Decimal is deprecated since v1.2.0. Use
        decimal.Decimal instead.`

    Example: Renaming
        When the symbol was also renamed on the way out, `new_name` gives the name
        it has now:

        ```python
        from typing import TYPE_CHECKING, TypeAlias

        from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

        if TYPE_CHECKING:
            from fractions import Fraction as _Fraction

            Rational: TypeAlias = _Fraction
        else:
            __getattr__ = deprecated_aliases(
                __name__,
                DeprecatedAlias(
                    "Rational",
                    new_module="fractions",
                    new_name="Fraction",
                    since="v1.3.0",
                ),
            )
        ```

        Without `new_module`, the symbol was renamed in this very module:

        ```python
        from typing import TYPE_CHECKING, TypeAlias

        from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases


        class Widget:
            ...  # Called Gadget before v1.4.0


        if TYPE_CHECKING:
            Gadget: TypeAlias = Widget
        else:
            __getattr__ = deprecated_aliases(
                __name__,
                DeprecatedAlias("Gadget", new_name="Widget", since="v1.4.0"),
            )
        ```

    Example: Custom message
        When the standard wording is not enough, an alias can give its whole
        message instead of `since`, as a template that gets the fully qualified old
        and new names as `{old}` and `{new}`:

        ```python
        from typing import TYPE_CHECKING, TypeAlias

        from frequenz.core.warnings import DeprecatedAlias, deprecated_aliases

        if TYPE_CHECKING:
            from decimal import Decimal as _Decimal

            Decimal: TypeAlias = _Decimal
        else:
            __getattr__ = deprecated_aliases(
                __name__,
                DeprecatedAlias(
                    "Decimal",
                    new_module="decimal",
                    message="{old} is deprecated since v1.2.0 and will be removed "
                    "in v2.0.0. Use {new} instead.",
                ),
            )
        ```

    Tip: Deprecating whole modules
        This function can't be used to deprecate a whole module. If you need to
        do that, you can keep a real `__init__.py` at the old path, with a
        `__getattr__ = deprecated_aliases(...)` that includes all the symbols
        that used to be reachable there.

    Warning: Security Warning
        This function imports the modules the aliases point at, so the aliases are
        as trusted as an `import` statement in this module. Write them out as
        literals; don't build them from anything that comes from outside the
        program.

    Args:
        module: The fully qualified name of the module defining the aliases.
        *aliases: The deprecated aliases, at least one, each with its own name.
        category: The category of the warning to emit.
        stacklevel: How far up the stack the warning is reported, counting from the
            `__getattr__` itself. The default of 2 points at the code reaching for
            the alias, and only needs raising if something wraps the returned
            function.

    Returns:
        A function suitable for use as the module's `__getattr__`.

    Raises:
        TypeError: If an argument has the wrong type, including an alias that is
            not a [`DeprecatedAlias`][..DeprecatedAlias]. Also later, when an alias
            is reached, if it resolves to a module rather than to a symbol in one.
        ValueError: If there are no aliases, if two of them have the same name, if
            one points at another alias in the same module, itself included, or if
            `stacklevel` is smaller than 1.
    """
    if not isinstance(module, str):
        raise TypeError(f"module must be a str, got {module!r}")
    if not (isinstance(category, type) and issubclass(category, Warning)):
        raise TypeError(f"category must be a Warning subclass, got {category!r}")
    if not isinstance(stacklevel, int):
        raise TypeError(f"stacklevel must be an int, got {stacklevel!r}")
    if stacklevel < 1:
        raise ValueError(f"stacklevel must be 1 or more, got {stacklevel!r}")
    if not aliases:
        raise ValueError(f"no aliases given for module {module!r}")
    by_name: dict[str, DeprecatedAlias] = {}
    for alias in aliases:
        # Also what a mapping, the form this function used to take, runs into.
        if not isinstance(alias, DeprecatedAlias):
            raise TypeError(f"aliases must be DeprecatedAlias instances, got {alias!r}")
        # A dict literal would keep the last of two entries silently; this is the
        # one place that can tell.
        if alias.name in by_name:
            raise ValueError(f"alias {alias.name!r} is given more than once")
        by_name[alias.name] = alias
    for alias in aliases:
        # Reaching it would call this `__getattr__` again, and never stop if the
        # aliases point at each other.
        if alias.new_module in (None, module) and alias.new_name in by_name:
            raise ValueError(
                f"the alias {alias.name!r} points at {module}.{alias.new_name}, "
                "which is a deprecated alias too"
            )

    def module_getattr(name: str) -> Any:
        """Return a deprecated alias, warning about its new location.

        Args:
            name: The name being looked up in the module.

        Returns:
            The aliased object.

        Raises:
            AttributeError: If the name is not one of the deprecated aliases.
            TypeError: If the alias resolves to a module, which this can't keep
                reachable from its old path.
        """
        alias = by_name.get(name)
        if alias is None:
            raise AttributeError(f"module {module!r} has no attribute {name!r}")
        new_module = module if alias.new_module is None else alias.new_module
        new = f"{new_module}.{alias.new_name}"
        # Resolved before the warning is raised, so an alias that can't work says
        # so instead of warning about a move that didn't happen. It also keeps the
        # `TypeError` below from being masked by an "error" filter turning that
        # warning into an exception first.
        value = getattr(importlib.import_module(new_module), alias.new_name)
        if isinstance(value, ModuleType):
            raise TypeError(
                f"the target of {module}.{name} is the module {new}, and this "
                "aliases names, not modules: the import system never consults a "
                f"package's __getattr__, so `import {module}.{name}` would fail "
                f"anyway. Keep a real __init__.py at {module}.{name}, with the "
                "aliases for the names it used to hold in it."
            )
        warnings.warn(
            alias.format_message(old=f"{module}.{name}", new=new),
            category,
            stacklevel=stacklevel,
        )
        return value

    return module_getattr

frequenz.core.warnings.ignoring_deprecations

ignoring_deprecations(
    *, message: str = "", module: str = ""
) -> ignoring_warnings

Ignore the deprecation warnings raised inside the block.

This is ignoring_warnings for the most common case, library code that has to touch a symbol it deprecated itself. The user was already warned by the deprecated symbol they used; warning them again from its internals, about something they can do nothing about, is noise.

All limitations of ignoring_warnings() apply

Read the documentation of ignoring_warnings, many limitations apply here too, as this is only a thin wrapper around it.

Tip

Wrap the call that reaches the deprecated symbol, not everything around it. A block that covers more than that also silences deprecations that have nothing to do with the one it was added for.

Example

import warnings

from frequenz.core.warnings import ignoring_deprecations


# A type that used to be public and is now deprecated.
class Wrapper:
    def __init__(self, raw: str) -> None:
        warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2)
        self.raw = raw


def from_wire(raw: str) -> Wrapper:
    # Whatever else this function does, the deprecations raised out there
    # are the user's business, so they stay outside the block.
    with ignoring_deprecations():
        return Wrapper(raw)

Inside a function that is itself deprecated

A deprecated function has already warned its caller about this code path, so the deprecations it reaches on the way are noise too. Even there, keep the block around the calls that raise them rather than putting it around the whole body, which this module can't do for you: it is a context manager and refuses to be used as a decorator.

import warnings

from typing_extensions import deprecated

from frequenz.core.warnings import ignoring_deprecations


# A type that used to be public and is now deprecated.
class Wrapper:
    def __init__(self, raw: str) -> None:
        warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2)
        self.raw = raw


async def save(wrapper: Wrapper) -> None:
    print(f"saving {wrapper.raw}")


@deprecated("Use from_wire() instead")
async def parse(raw: str) -> Wrapper:
    with ignoring_deprecations():
        wrapper = Wrapper(raw)
    await save(wrapper)  # Outside: the scope note applies across an await.
    return wrapper

PARAMETER	DESCRIPTION
message	A regular expression the start of the warning message must match, case insensitively. The default matches every deprecation, which is usually right: the way to be precise here is a short block, not a narrow filter. Use it when the call being wrapped can also raise a deprecation that should be heard. TYPE: str DEFAULT: ''
module	A regular expression the start of the module name must match. The default matches every module. TYPE: str DEFAULT: ''

RETURNS	DESCRIPTION
ignoring_warnings	A context manager that ignores deprecation warnings.

RAISES	DESCRIPTION
TypeError	If any argument has the wrong type.
error	If message or module is not a valid regular expression.

Source code in src/frequenz/core/warnings/_ignoring.py

def ignoring_deprecations(  # noqa: DOC502
    *, message: str = "", module: str = ""
) -> ignoring_warnings:
    """Ignore the deprecation warnings raised inside the block.

    This is [`ignoring_warnings`][..ignoring_warnings] for the most common case,
    library code that has to touch a symbol it deprecated itself. The user was
    already warned by the deprecated symbol they used; warning them again from its
    internals, about something they can do nothing about, is noise.

    Warning: All limitations of `ignoring_warnings()` apply
        Read the documentation of [`ignoring_warnings`][..ignoring_warnings],
        many limitations apply here too, as this is only a thin wrapper around it.

    Tip:
        Wrap the call that reaches the deprecated symbol, not everything around
        it. A block that covers more than that also silences deprecations that have
        nothing to do with the one it was added for.

    Example:
        ```python
        import warnings

        from frequenz.core.warnings import ignoring_deprecations


        # A type that used to be public and is now deprecated.
        class Wrapper:
            def __init__(self, raw: str) -> None:
                warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2)
                self.raw = raw


        def from_wire(raw: str) -> Wrapper:
            # Whatever else this function does, the deprecations raised out there
            # are the user's business, so they stay outside the block.
            with ignoring_deprecations():
                return Wrapper(raw)
        ```

    Example: Inside a function that is itself deprecated
        A deprecated function has already warned its caller about this code path, so
        the deprecations it reaches on the way are noise too. Even there, keep the
        block around the calls that raise them rather than putting it around the
        whole body, which this module can't do for you: it is a context manager and
        refuses to be used as a decorator.

        ```python
        import warnings

        from typing_extensions import deprecated

        from frequenz.core.warnings import ignoring_deprecations


        # A type that used to be public and is now deprecated.
        class Wrapper:
            def __init__(self, raw: str) -> None:
                warnings.warn("Wrapper is deprecated", DeprecationWarning, stacklevel=2)
                self.raw = raw


        async def save(wrapper: Wrapper) -> None:
            print(f"saving {wrapper.raw}")


        @deprecated("Use from_wire() instead")
        async def parse(raw: str) -> Wrapper:
            with ignoring_deprecations():
                wrapper = Wrapper(raw)
            await save(wrapper)  # Outside: the scope note applies across an await.
            return wrapper
        ```

    Args:
        message: A regular expression the start of the warning message must match,
            case insensitively. The default matches every deprecation, which is
            usually right: the way to be precise here is a short block, not a narrow
            filter. Use it when the call being wrapped can also raise a deprecation
            that should be heard.
        module: A regular expression the start of the module name must match. The
            default matches every module.

    Returns:
        A context manager that ignores deprecation warnings.

    Raises:
        TypeError: If any argument has the wrong type.
        re.error: If `message` or `module` is not a valid regular expression.
    """
    return ignoring_warnings(
        category=DeprecationWarning, message=message, module=module
    )