Decorators

decorators

Decorators for message classes.

Provides @message_dataclass (and its aliases @event_dataclass,
@command_dataclass, @query_dataclass) to reduce boilerplate when
defining message types. The decorated class is a frozen dataclass whose
fields are automatically exposed via get_payload_fields() and are used
by from_payload_fields() for reconstruction.

Example

class OrderPayload:
    def __init__(self, order_id: str, customer_id: str, total: float) -> None:
        self.order_id = order_id
        self.customer_id = customer_id
        self.total = total


class Event[T]:
    # Inline stub for the example.
    pass


@event_dataclass
class OrderCreated(Event[OrderPayload]):
    order_id: str
    customer_id: str
    total: float


event = OrderCreated(order_id="ORD-001", customer_id="CUST-42", total=99.95)

event_dataclass = message_dataclass module-attribute

Alias for @message_dataclass intended for domain events.

command_dataclass = message_dataclass module-attribute

Alias for @message_dataclass intended for commands.

query_dataclass = message_dataclass module-attribute

Alias for @message_dataclass intended for queries.

message_dataclass(cls: type[_M] | None = None) -> type[_M] | Callable[[type[_M]], type[_M]]

message_dataclass(cls: type[_M]) -> type[_M]

message_dataclass(
    cls: None = None,
) -> Callable[[type[_M]], type[_M]]

Decorate a class as a message dataclass.

The decorator applies @dataclass(frozen=False) and then replaces
__setattr__ with a custom implementation that raises
FrozenInstanceError after __init__ completes. It also
patches get_payload_fields and from_payload_fields onto the
class so that payload data is automatically derived from its fields.

When the decorated class inherits from an abstract base (e.g.
Event, Command, Query), the decorator
automatically patches _payload, value, and
from_payload_fields — and removes them from __abstractmethods__
— so the concrete subclass is instantiable without manual stubs.

Parameters:

Name	Type	Description	Default
cls	type[_M] | None	The class to decorate (when used without arguments).	None

Returns:

Type	Description
type[_M] | Callable[[type[_M]], type[_M]]	The decorated class (or a wrapper when called with keyword arguments).

Source code in src/forging_blocks/domain/messages/decorators.py

def message_dataclass(
    cls: type[_M] | None = None,
) -> type[_M] | Callable[[type[_M]], type[_M]]:
    """Decorate a class as a message dataclass.

    The decorator applies ``@dataclass(frozen=False)`` and then replaces
    ``__setattr__`` with a custom implementation that raises
    ``FrozenInstanceError`` after ``__init__`` completes.  It also
    patches ``get_payload_fields`` and ``from_payload_fields`` onto the
    class so that payload data is automatically derived from its fields.

    When the decorated class inherits from an abstract base (e.g.
    `Event`, `Command`, `Query`), the decorator
    automatically patches ``_payload``, ``value``, and
    ``from_payload_fields`` — and removes them from ``__abstractmethods__``
    — so the concrete subclass is instantiable without manual stubs.

    Args:
        cls: The class to decorate (when used without arguments).

    Returns:
        The decorated class (or a wrapper when called with keyword arguments).
    """

    def wrap(cls: type[_M]) -> type[_M]:
        dc_cls: type[_M] = dataclass(frozen=False, eq=False)(cls)

        original_init = dc_cls.__init__

        def frozen_setattr(self: object, name: str, value: object) -> None:
            """Raise FrozenInstanceError for attribute assignment after init."""
            if getattr(self, "__init_finished__", False):
                raise dataclasses.FrozenInstanceError(f"cannot assign to field {name!r}")
            object.__setattr__(self, name, value)

        def new_init(
            self: _M,
            *args: Any,
            metadata: MessageMetadata | None = None,
            **kwargs: Any,
        ) -> None:
            object.__setattr__(self, "__init_finished__", False)
            original_init(self, *args, **kwargs)
            effective_type = type(self).__name__
            object.__setattr__(
                self, "_metadata", metadata or MessageMetadata(message_type=effective_type)
            )
            object.__setattr__(self, "__init_finished__", True)

        dc_cls.__setattr__ = frozen_setattr

        def get_payload_fields(self: _M) -> dict[str, object]:
            return {
                name: getattr(self, name)
                for name in cast(Any, dc_cls).__dataclass_fields__
                if not name.startswith("_") and name not in ("metadata",)
            }

        @classmethod
        def from_payload_fields(
            cls: type[_M],
            data: dict[str, object],
            metadata: MessageMetadata,
        ) -> _M:
            fields = cast(Any, dc_cls).__dataclass_fields__
            unknown = data.keys() - fields.keys()
            if unknown:
                raise TypeError(
                    f"Unknown field(s) in payload for {cls.__name__}: {sorted(unknown)}"
                )
            return cls(metadata=metadata, **data)

        patched = cast(Any, dc_cls)
        patched.__init__ = new_init
        patched.get_payload_fields = get_payload_fields
        patched.from_payload_fields = from_payload_fields

        abstract_methods: frozenset[str] = getattr(dc_cls, "__abstractmethods__", frozenset())
        if "_payload" in abstract_methods:
            patched._payload = property(lambda self: self.get_payload_fields())
            dc_cls.__abstractmethods__ = frozenset(
                m for m in dc_cls.__abstractmethods__ if m != "_payload"
            )
        if "value" in abstract_methods:
            patched.value = property(lambda self: self.get_payload_fields())
            dc_cls.__abstractmethods__ = frozenset(
                m for m in dc_cls.__abstractmethods__ if m != "value"
            )
        if "from_payload_fields" in abstract_methods:
            dc_cls.__abstractmethods__ = frozenset(
                m for m in dc_cls.__abstractmethods__ if m != "from_payload_fields"
            )

        if not isinstance(dc_cls, _PatchedMessage):
            raise TypeError(
                f"{dc_cls.__name__!r} does not satisfy _PatchedMessage after decoration. "
                "This is a bug in message_dataclass."
            )

        return cast(type[_M], dc_cls)

    return wrap if cls is None else wrap(cls)