Skip to content

WrappedAggregate

Bases: Generic[TAggregate]

Wraps an aggregate instance with stream-level metadata.

Provides access to the aggregate alongside information such as version, timestamps, and whether the aggregate is newly created. Follows the same wrapper pattern as WrappedEvent.

Attributes:

Name Type Description
aggregate TAggregate

The aggregate instance.
stream_id StreamId

The stream identity (UUID + category).
context Context

The context passed for this operation, if any.
created_at datetime | None

Timestamp of the first event in the stream.
updated_at datetime | None

Timestamp of the last event in the stream.
stored_version int

Number of events persisted before this session.
Source code in event_sourcery/event_sourcing/aggregate.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
@dataclasses.dataclass()
class WrappedAggregate(Generic[TAggregate]):
    """
    Wraps an aggregate instance with stream-level metadata.

    Provides access to the aggregate alongside information such as version,
    timestamps, and whether the aggregate is newly created. Follows the same
    wrapper pattern as ``WrappedEvent``.

    Attributes:
        aggregate: The aggregate instance.
        stream_id: The stream identity (UUID + category).
        context: The context passed for this operation, if any.
        created_at: Timestamp of the first event in the stream.
        updated_at: Timestamp of the last event in the stream.
        stored_version: Number of events persisted before this session.
    """

    aggregate: TAggregate
    stream_id: StreamId
    context: Context = dataclasses.field(default_factory=Context)
    stored_version: int = 0
    created_at: datetime | None = None
    updated_at: datetime | None = None

    @property
    def version(self) -> int:
        """Current version of the aggregate, including pending events."""
        return self.stored_version + len(getattr(self.aggregate, "__changes__", []))

    @property
    def is_new(self) -> bool:
        """Whether the aggregate is newly created (no events existed before)."""
        return self.stored_version == 0

    def get_context(self, context_type: type[TContext]) -> TContext:
        """Convert the stored context to a specific context type.

        Args:
            context_type: The target context class to validate against.

        Returns:
            An instance of *context_type* populated from the stored context data.
        """
        return context_type.model_validate(self.context.model_dump())

is_new property

Whether the aggregate is newly created (no events existed before).

version property

Current version of the aggregate, including pending events.

get_context(context_type)

Convert the stored context to a specific context type.

Parameters:

Name Type Description Default
context_type type[TContext]

The target context class to validate against.

 required

Returns:

Type Description
TContext

An instance of context_type populated from the stored context data.
Source code in event_sourcery/event_sourcing/aggregate.py 
114
115
116
117
118
119
120
121
122
123
def get_context(self, context_type: type[TContext]) -> TContext:
    """Convert the stored context to a specific context type.

    Args:
        context_type: The target context class to validate against.

    Returns:
        An instance of *context_type* populated from the stored context data.
    """
    return context_type.model_validate(self.context.model_dump())