Command

command

Module defining the Command base class for commands.

Command

Bases: Message[RawCommandType]

Base class for all commands.

Commands represent an intent to do something in the system.
They are requests that may succeed or fail, and are handled by a command handler.

Commands are named in imperative mood (e.g., CreateOrder, RegisterCustomer,
ProcessPayment).

Example

class CreateOrderPayload:
    def __init__(self, customer_id: str, items: list) -> None:
        self.customer_id = customer_id
        self.items = items


class CreateOrder(Command[CreateOrderPayload]):
    def __init__(
        self, customer_id: str, items: list, metadata: MessageMetadata | None = None
    ):
        super().__init__(metadata)
        self._customer_id = customer_id
        self._items = items

    @property
    def _payload(self) -> CreateOrderPayload:
        return CreateOrderPayload(
            customer_id=self._customer_id,
            items=self._items,
        )

    @classmethod
    def from_payload_fields(
        cls, data: CreateOrderPayload, metadata: MessageMetadata
    ) -> "CreateOrder":
        return cls(
            customer_id=data.customer_id,
            items=data.items,
            metadata=metadata,
        )

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

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

class Command[RawCommandType](Message[RawCommandType]):
    """Base class for all commands.

    Commands represent an intent to do something in the system.
    They are requests that may succeed or fail, and are handled by a command handler.

    Commands are named in imperative mood (e.g., CreateOrder, RegisterCustomer,
    ProcessPayment).


    Example:
        ```python
        class CreateOrderPayload:
            def __init__(self, customer_id: str, items: list) -> None:
                self.customer_id = customer_id
                self.items = items


        class CreateOrder(Command[CreateOrderPayload]):
            def __init__(
                self, customer_id: str, items: list, metadata: MessageMetadata | None = None
            ):
                super().__init__(metadata)
                self._customer_id = customer_id
                self._items = items

            @property
            def _payload(self) -> CreateOrderPayload:
                return CreateOrderPayload(
                    customer_id=self._customer_id,
                    items=self._items,
                )

            @classmethod
            def from_payload_fields(
                cls, data: CreateOrderPayload, metadata: MessageMetadata
            ) -> "CreateOrder":
                return cls(
                    customer_id=data.customer_id,
                    items=data.items,
                    metadata=metadata,
                )

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

    """

    @property
    def command_id(self) -> UUID:
        """Get the unique identifier for this command (same as message_id)."""
        return self.message_id

    @property
    def issued_at(self) -> datetime:
        """Get the timestamp when this command was issued (same as created_at)."""
        return self.created_at

command_id: UUID property

Get the unique identifier for this command (same as message_id).

issued_at: datetime property

Get the timestamp when this command was issued (same as created_at).