Skip to content

Event

event

Module defining the base Event class for events.

Event

Bases: Message[RawEventType]

Base class for all events.

Events represent something significant that happened in the system.
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 OrderCreatedPayload:
    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 OrderCreated(Event[OrderCreatedPayload]):
    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) -> OrderCreatedPayload:
        return OrderCreatedPayload(
            order_id=self._order_id,
            customer_id=self._customer_id,
            total=self._total,
        )

    @classmethod
    def from_payload_fields(
        cls, data: OrderCreatedPayload, metadata: MessageMetadata
    ) -> "OrderCreated":
        return cls(
            order_id=data.order_id,
            customer_id=data.customer_id,
            total=data.total,
        )

    @property
    def value(self) -> OrderCreatedPayload:
        return self._payload
Source code in src/forging_blocks/domain/messages/event.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
class Event[RawEventType](Message[RawEventType]):
    """Base class for all events.

    Events represent something significant that happened in the system.
    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 OrderCreatedPayload:
            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 OrderCreated(Event[OrderCreatedPayload]):
            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) -> OrderCreatedPayload:
                return OrderCreatedPayload(
                    order_id=self._order_id,
                    customer_id=self._customer_id,
                    total=self._total,
                )

            @classmethod
            def from_payload_fields(
                cls, data: OrderCreatedPayload, metadata: MessageMetadata
            ) -> "OrderCreated":
                return cls(
                    order_id=data.order_id,
                    customer_id=data.customer_id,
                    total=data.total,
                )

            @property
            def value(self) -> OrderCreatedPayload:
                return self._payload
        ```

    """

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

    @property
    @abstractmethod
    def _payload(self) -> RawEventType:
        """Return the event-specific payload data.

        Subclasses MUST implement this property to return the event-specific
        data.
        """

occurred_at: datetime property

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