Skip to content

Event

event

Module defining the base Event class for domain events.

Event

Bases: Message[RawEventType]

Base class for all domain events.

Domain events represent something significant that happened in the domain.
They are immutable facts about the past that other parts of the system can react to.

Events are named in past tense (e.g., OrderCreated, CustomerRegistered,
PaymentProcessed).

Example
class OrderCreated(Event[dict[str, object]]):
    def __init__(self, order_id: str, customer_id: str, total: float):
        super().__init__()
        self._order_id = order_id
        self._customer_id = customer_id
        self._total = total

    @property
    def _payload(self) -> dict[str, object]:
        return {
            "order_id": self._order_id,
            "customer_id": self._customer_id,
            "total": self._total,
        }
Source code in src/forging_blocks/foundation/messages/event.py
class Event[RawEventType](Message[RawEventType]):
    """Base class for all domain events.

    Domain events represent something significant that happened in the domain.
    They are immutable facts about the past that other parts of the system can react to.

    Events are named in past tense (e.g., OrderCreated, CustomerRegistered,
    PaymentProcessed).

    Example:
        ```python
        class OrderCreated(Event[dict[str, object]]):
            def __init__(self, order_id: str, customer_id: str, total: float):
                super().__init__()
                self._order_id = order_id
                self._customer_id = customer_id
                self._total = total

            @property
            def _payload(self) -> dict[str, object]:
                return {
                    "order_id": self._order_id,
                    "customer_id": self._customer_id,
                    "total": self._total,
                }
        ```
    """

    @property
    def occurred_at(self) -> datetime:
        """Get the timestamp when this event occurred (UTC timezone)."""
        return self.created_at

    @property
    @abstractmethod
    def _payload(self) -> dict[str, object]: ...

occurred_at: datetime property

Get the timestamp when this event occurred (UTC timezone).