Messages

Commands and events are the two message types in Protean. Both are immutable DTOs that carry data and tracing metadata.

Guides: Commands · Events · Raising Events

---

BaseCommand

BaseCommand(*args: Any, **kwargs: Any)

Bases: BaseMessageType

Base class for domain commands -- immutable DTOs representing an intent to change aggregate state.

Commands are named with imperative verbs (PlaceOrder, RegisterUser, CancelReservation) and processed by command handlers. They are immutable after construction and carry metadata for tracing (correlation ID, causation ID, origin stream).

Fields are declared using standard Python type annotations with optional Field constraints.

Meta Options

Option	Type	Description
part_of	type	The aggregate class this command targets. Required.

Create a new command instance.

Accepts keyword arguments matching the declared fields. Optionally, a dict can be passed as a positional argument to serve as a template — keyword arguments take precedence over template values.

PARAMETER	DESCRIPTION
*args	Optional template dictionaries for field values. TYPE: dict DEFAULT: ()
**kwargs	Field values for the command. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
ValidationError	If field validation fails.

Example::

# Keyword arguments
PlaceOrder(order_id="abc", amount=100)

# Template dict pattern
PlaceOrder({"order_id": "abc"}, amount=100)

Source code in src/protean/core/command.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Create a new command instance.

    Accepts keyword arguments matching the declared fields. Optionally,
    a ``dict`` can be passed as a positional argument to serve as a
    template — keyword arguments take precedence over template values.

    Args:
        *args (dict): Optional template dictionaries for field values.
        **kwargs (Any): Field values for the command.

    Raises:
        ValidationError: If field validation fails.

    Example::

        # Keyword arguments
        PlaceOrder(order_id="abc", amount=100)

        # Template dict pattern
        PlaceOrder({"order_id": "abc"}, amount=100)
    """
    incoming_metadata = kwargs.pop("_metadata", None)

    # Support template dict pattern: Command({"key": "val"}, key2="val2")
    # Keyword args take precedence over template dict values.
    if args:
        merged: dict[str, Any] = {}
        for template in args:
            if not isinstance(template, dict):
                raise AssertionError(
                    f"Positional argument {template} passed must be a dict. "
                    f"This argument serves as a template for loading common "
                    f"values.",
                )
            merged.update(template)
        merged.update(kwargs)
        kwargs = merged

    # Template dicts (e.g. from to_dict()) may re-introduce _metadata
    # and _version; prefer the explicitly passed keyword arg, fall back
    # to template value.  _version is an aggregate-internal field and is
    # not part of the command schema, so discard it silently.
    template_metadata = kwargs.pop("_metadata", None)
    if incoming_metadata is None:
        incoming_metadata = template_metadata
    kwargs.pop("_version", None)

    try:
        super().__init__(**kwargs)
    except PydanticValidationError as e:
        raise ValidationError(convert_pydantic_errors(e))

    # Build metadata
    self._build_metadata(incoming_metadata)

    object.__setattr__(self, "_initialized", True)

---

BaseEvent

BaseEvent(*args: Any, **kwargs: Any)

Bases: BaseMessageType

Base class for domain events -- immutable facts representing state changes that have occurred in the domain.

Events are named in past tense (OrderPlaced, CustomerRegistered, PaymentConfirmed) and enable decoupled communication between system components. They are immutable after construction and carry metadata for tracing (correlation ID, causation ID, stream position).

Fields are declared using standard Python type annotations with optional Field constraints.

Meta Options

Option	Type	Description
part_of	type	The aggregate class that raises this event. Required.

Create a new event instance.

Accepts keyword arguments matching the declared fields. Optionally, a dict can be passed as a positional argument to serve as a template — keyword arguments take precedence over template values.

Events are typically created by calling self.raise_() inside an aggregate method rather than being instantiated directly.

PARAMETER	DESCRIPTION
*args	Optional template dictionaries for field values. TYPE: dict DEFAULT: ()
**kwargs	Field values for the event. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
ValidationError	If field validation fails.

Example::

# Raised from an aggregate method
self.raise_(OrderPlaced(order_id=self.id, amount=self.total))

# Template dict pattern (e.g. during reconstitution)
OrderPlaced({"order_id": "abc"}, amount=100)

Source code in src/protean/core/event.py

def __init__(self, *args: Any, **kwargs: Any) -> None:
    """Create a new event instance.

    Accepts keyword arguments matching the declared fields. Optionally,
    a ``dict`` can be passed as a positional argument to serve as a
    template — keyword arguments take precedence over template values.

    Events are typically created by calling ``self.raise_()`` inside an
    aggregate method rather than being instantiated directly.

    Args:
        *args (dict): Optional template dictionaries for field values.
        **kwargs (Any): Field values for the event.

    Raises:
        ValidationError: If field validation fails.

    Example::

        # Raised from an aggregate method
        self.raise_(OrderPlaced(order_id=self.id, amount=self.total))

        # Template dict pattern (e.g. during reconstitution)
        OrderPlaced({"order_id": "abc"}, amount=100)
    """
    incoming_metadata = kwargs.pop("_metadata", None)
    expected_version = kwargs.pop("_expected_version", -1)

    # Support template dict pattern: Event({"key": "val"}, key2="val2")
    # Keyword args take precedence over template dict values.
    if args:
        merged: dict[str, Any] = {}
        for template in args:
            if not isinstance(template, dict):
                raise AssertionError(
                    f"Positional argument {template} passed must be a dict. "
                    f"This argument serves as a template for loading common "
                    f"values.",
                )
            merged.update(template)
        merged.update(kwargs)
        kwargs = merged

    # Template dicts (e.g. from to_dict()) may re-introduce _metadata,
    # _expected_version, and _version; prefer the explicitly passed
    # keyword args.  _version is an aggregate-internal field and is not
    # part of the event schema, so discard it silently.
    template_metadata = kwargs.pop("_metadata", None)
    if incoming_metadata is None:
        incoming_metadata = template_metadata
    template_expected_version = kwargs.pop("_expected_version", None)
    if expected_version == -1 and template_expected_version is not None:
        expected_version = template_expected_version
    kwargs.pop("_version", None)

    try:
        super().__init__(**kwargs)
    except PydanticValidationError as e:
        raise ValidationError(convert_pydantic_errors(e))

    # Store expected version as regular attr (before _initialized is set)
    object.__setattr__(self, "_expected_version", expected_version)

    # Build metadata
    self._build_metadata(incoming_metadata)

    object.__setattr__(self, "_initialized", True)