BaseAggregate

Base class for aggregate root entities, the primary building block for modeling domain concepts that enforce consistency rules and manage state changes.

See Aggregates guide for practical usage and Aggregates concept for design rationale.

Bases: BaseEntity

Base class for aggregate root entities -- the primary building block for modeling domain concepts.

Aggregates enforce consistency rules and define transaction boundaries. They inherit all entity capabilities (fields, identity, invariants) and add versioning for optimistic concurrency, event raising via raise_(), and event-sourcing support via _apply() / from_events().

Meta Options

Option	Type	Description
event_sourced	bool	Enable event-sourcing mode (default: False).
fact_events	bool	Auto-generate fact events on persistence (default: False).
stream_category	str	Override the event stream category name.
provider	str	The persistence provider name (default: "default").
schema_name	str	The storage table/collection name.
auto_add_id_field	bool	Whether to auto-inject an id field (default: True).
reserved	tuple[str, ...]	Field names that once existed and must never be reused. Removing a field from an event-sourced aggregate is safe only when its name is reserved. During replay, an assignment to a reserved name is dropped, and so is a call to its add_<name>/remove_<name> helper. Child entities those calls would have added are dropped with them.

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

def __init__(self, *args: Any, **kwargs: Any) -> None:
    # Pop _version before Pydantic init (it's a PrivateAttr,
    # and extra="forbid" would reject it). Restore after construction.
    # Check both kwargs and positional dict args (template dict pattern).
    version = kwargs.pop("_version", None)
    if version is None:
        for arg in args:
            if isinstance(arg, dict) and "_version" in arg:
                version = arg.pop("_version")
                break
    if version is None:
        version = -1

    super().__init__(*args, **kwargs)

    # Restore _version from kwargs or default
    self._version = version

    # Set self as root and owner
    self._set_root_and_owner(self, self)

    # Increment version and set next version
    self._next_version = self._version + 1

raise_

raise_(event: Any) -> None

Raise a domain event on this aggregate.

Enriches the event with metadata (identity, stream, sequence, checksum) and appends it to self._events.

On an event-sourced aggregate, the event's @apply handler then runs. If any step raises (an enricher, the handler, or an invariant check around it), _events, _version and _event_position go back to their values from before the call and the error propagates. When an invariant check fails, atomic_change also undoes the field changes the handler made. When the handler raises its own error, its field changes stay.

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

def raise_(self, event: Any) -> None:
    """Raise a domain event on this aggregate.

    Enriches the event with metadata (identity, stream, sequence,
    checksum) and appends it to ``self._events``.

    On an event-sourced aggregate, the event's ``@apply`` handler then
    runs. If any step raises (an enricher, the handler, or an invariant
    check around it), ``_events``, ``_version`` and ``_event_position``
    go back to their values from before the call and the error
    propagates. When an invariant check fails, ``atomic_change`` also
    undoes the field changes the handler made. When the handler raises
    its own error, its field changes stay.
    """
    # Guard: temporal aggregates are read-only
    if self._is_temporal:
        raise IncorrectUsageError(
            "Cannot raise events on a temporally-loaded aggregate. "
            "Temporal aggregates are read-only."
        )

    # Verify that event is associated with this aggregate
    if event.meta_.part_of != self.__class__:
        raise ConfigurationError(
            f"Event `{event.__class__.__name__}` is not associated with"
            f" aggregate `{self.__class__.__name__}`"
        )

    # Warn once per type when a deprecated event is raised.
    self._warn_if_deprecated(event.__class__)

    id_field_name = getattr(self.__class__, _ID_FIELD_NAME, None)
    identifier = getattr(self, id_field_name) if id_field_name else None

    # Set Fact Event stream to be `<aggregate_stream_name>-fact`
    if event.__class__.meta_.is_fact_event:
        stream = f"{self.meta_.stream_category}-fact-{identifier}"
    else:
        stream = f"{self.meta_.stream_category}-{identifier}"

    # Anything below can raise: an enricher, the metadata build, the
    # @apply handler or an invariant check around it.
    mark = self._event_mark()
    try:
        self._record_event(event, stream)
    except BaseException:
        self._rewind_events(mark)
        raise

from_events classmethod

from_events(events: list[Any]) -> BaseAggregate

Event-Sourcing: reconstruct an aggregate from a list of events.

Creates a blank aggregate via _create_for_reconstitution() and applies all events uniformly through _apply(). The first event's @apply handler must set ALL fields including identity.

RAISES	DESCRIPTION
IncorrectUsageError	If events is empty. An aggregate cannot be reconstructed without at least one event.

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

@classmethod
def from_events(cls, events: list[Any]) -> "BaseAggregate":
    """Event-Sourcing: reconstruct an aggregate from a list of events.

    Creates a blank aggregate via ``_create_for_reconstitution()`` and
    applies all events uniformly through ``_apply()``.  The first event's
    ``@apply`` handler must set ALL fields including identity.

    Raises:
        IncorrectUsageError: If ``events`` is empty. An aggregate cannot
            be reconstructed without at least one event.
    """
    if not events:
        raise IncorrectUsageError(
            f"Cannot reconstitute `{cls.__name__}` from an empty event list"
        )

    aggregate = cls._create_for_reconstitution()

    for event in events:
        aggregate._apply(event)

    aggregate._disable_invariant_checks = False
    return aggregate