Exceptions with Template-based messages

README.md

Inspired by https://github.com/pyctrl/izulu

But super minimal to be used just as a copy-pasted module.

error.py

"""Minimal templated exception helper."""

from string import Formatter


class Error(Exception):
    """Base templated exception class."""

    __template__ = "Exception occurred"

    @classmethod
    def _collect_annotations(cls) -> dict[str, object]:
        """Collect all annotated fields from the class hierarchy."""
        annotations: dict[str, object] = {}
        for class_ in reversed(cls.__mro__):
            annotations.update(getattr(class_, "__annotations__", {}))
        return annotations

    @classmethod
    def _format_fields(cls, names: set[str]) -> str:
        """Format field names in a deterministic error message."""
        return ", ".join(f"'{name}'" for name in sorted(names))

    @classmethod
    def _template_fields(cls, template: str) -> set[str]:
        """Extract plain field names used in a format template."""
        fields: set[str] = set()
        for _, field_name, _, _ in Formatter().parse(template):
            if not field_name:
                continue
            field = field_name.split(".", maxsplit=1)[0].split("[", maxsplit=1)[0]
            fields.add(field)
        return fields

    def __init__(self, **kwargs: object) -> None:
        annotations = self._collect_annotations()
        undeclared = set(kwargs) - set(annotations)
        if undeclared:
            raise TypeError(f"Undeclared arguments: {self._format_fields(undeclared)}")

        missing = {
            field
            for field in annotations
            if field not in kwargs and not hasattr(type(self), field)
        }
        if missing:
            raise TypeError(f"Missing arguments: {self._format_fields(missing)}")

        for key, value in kwargs.items():
            setattr(self, key, value)

        template = getattr(type(self), "__template__", self.__template__)
        missing_annotations = self._template_fields(template) - set(annotations)
        if missing_annotations:
            raise ValueError(
                f"Fields must be annotated: "
                f"{self._format_fields(missing_annotations)}",
            )

        payload = {field: getattr(self, field) for field in annotations}
        super().__init__(template.format(**payload))

    def __repr__(self) -> str:
        """Represent exception with all declared fields."""
        annotations = self._collect_annotations()
        module = type(self).__module__
        qualname = type(self).__qualname__
        if not annotations:
            return f"{module}.{qualname}()"
        args = ", ".join(f"{field}={getattr(self, field)!r}" for field in annotations)
        return f"{module}.{qualname}({args})"

LICENSE

MIT License

Copyright (c) 2026 mahenzon

Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:

The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.

THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

test_error.py

import pytest

from taskiq.error import Error
from taskiq.exceptions import SecurityError, TaskiqResultTimeoutError


class SimpleError(Error):
    __template__ = "simple"


class ValueTemplateError(Error):
    __template__ = "value={value}"
    value: int


class DefaultValueTemplateError(Error):
    __template__ = "value={value}"
    value: int = 10


class BaseError(Error):
    __template__ = "base={base}, child={child}"
    base: int = 1


class ChildError(BaseError):
    child: str


class MissingAnnotationError(Error):
    __template__ = "value={value}"


class IndexedTemplateError(Error):
    __template__ = "{payload[key]}"
    payload: dict[str, str]


def test_simple_error_message_and_repr() -> None:
    error = SimpleError()
    assert str(error) == "simple"
    assert error.args == ("simple",)
    assert repr(error).endswith(".SimpleError()")


def test_template_with_required_value() -> None:
    error = ValueTemplateError(value=3)
    assert str(error) == "value=3"
    assert repr(error).endswith(".ValueTemplateError(value=3)")


def test_missing_argument_raises_type_error() -> None:
    with pytest.raises(TypeError, match="Missing arguments: 'value'"):
        ValueTemplateError()


def test_undeclared_argument_raises_type_error() -> None:
    with pytest.raises(TypeError, match="Undeclared arguments: 'extra'"):
        ValueTemplateError(value=1, extra=2)


def test_default_value_is_used_without_kwargs() -> None:
    error = DefaultValueTemplateError()
    assert str(error) == "value=10"
    assert repr(error).endswith(".DefaultValueTemplateError(value=10)")


def test_annotations_are_collected_from_inheritance() -> None:
    error = ChildError(child="ok")
    assert str(error) == "base=1, child=ok"
    assert repr(error).endswith(".ChildError(base=1, child='ok')")


def test_template_fields_must_be_annotated() -> None:
    with pytest.raises(ValueError, match="Fields must be annotated: 'value'"):
        MissingAnnotationError()


def test_indexed_template_field_does_not_require_extra_annotation() -> None:
    error = IndexedTemplateError(payload={"key": "value"})
    assert str(error) == "value"


def test_taskiq_exceptions_use_error_base_correctly() -> None:
    timeout_error = TaskiqResultTimeoutError(timeout=1.5)
    security_error = SecurityError(description="boom")
    assert str(timeout_error) == "Waiting for task results has timed out, timeout=1.5"
    assert str(security_error) == "Security exception occurred: boom"