Skip to content

Aggregate Repository

aggregate_repository

Aggregate RepositoryPort implementation.

Provides a repository specifically designed for AggregateRoot persistence
with event sourcing support.

AggregateRepository

Bases: InMemoryRepository[TAggregateRoot, TId]

RepositoryPort for persisting event-sourced aggregates.
Coordinates event store writes with in-memory snapshot caching
for AggregateRoot subtypes.

Class Type Parameters:

Name Bound or Constraints Description Default
EventPayloadType

The event payload type tracked by the event store.
Flows through the public generic interface.

 required
TAggregateRoot AggregateRoot[UUID, Any]

An AggregateRoot subtype with UUID identity whose event
payload type must match EventPayloadType at each call site. The
bound uses Any as the second type argument — not because the
event type is untyped, but because PEP 695 (and the underlying type
system) forbids one TypeVar from appearing inside another TypeVar's
bound. The cast in save is the explicit, localized bridge
across this gap. The invariant is enforced by construction.

 required
TId UUID

The aggregate identity type, bounded by UUID.

 required
Example 
class Event[T]:
    pass


class AggregateRoot[TId, TPayload]:
    def __init__(self, aggregate_id: TId) -> None:
        self.id = aggregate_id


class InMemoryEventStore[T]:
    def __init__(self) -> None: ...

    async def append_events(
        self, aggregate_id: object, events: list[object], expected_version: int
    ) -> None: ...
    async def get_events(self, aggregate_id: object) -> list[object]: ...
    async def get_current_version(self, aggregate_id: object) -> int: ...


class MyAggregate(AggregateRoot[UUID, str]):
    def __init__(self, aggregate_id: UUID) -> None:
        super().__init__(aggregate_id)

    def _handle(self, event: Event[str]) -> None:
        pass


event_store = InMemoryEventStore[str]()
aggregate_id = UUID("00000000-0000-0000-0000-000000000001")
repo = AggregateRepository[str, MyAggregate, UUID](
    event_store=event_store,
    aggregate_type=MyAggregate,
)


async def main() -> None:
    aggregate = MyAggregate(aggregate_id)
    await repo.save(aggregate)
    retrieved = await repo.get_by_id(aggregate_id)
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
 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
 73
 74
 75
 76
 77
 78
 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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
class AggregateRepository[
    EventPayloadType,
    TAggregateRoot: AggregateRoot[UUID, Any],
    TId: UUID,
](InMemoryRepository[TAggregateRoot, TId]):
    """RepositoryPort for persisting event-sourced aggregates.
    Coordinates event store writes with in-memory snapshot caching
    for AggregateRoot subtypes.

    Type Parameters:
        EventPayloadType: The event payload type tracked by the event store.
            Flows through the public generic interface.
        TAggregateRoot: An AggregateRoot subtype with UUID identity whose event
            payload type must match ``EventPayloadType`` at each call site. The
            bound uses ``Any`` as the second type argument — not because the
            event type is untyped, but because PEP 695 (and the underlying type
            system) forbids one TypeVar from appearing inside another TypeVar's
            bound. The ``cast`` in `save` is the explicit, localized bridge
            across this gap. The invariant is enforced by construction.
        TId: The aggregate identity type, bounded by ``UUID``.

    Example:
        ```python
        class Event[T]:
            pass


        class AggregateRoot[TId, TPayload]:
            def __init__(self, aggregate_id: TId) -> None:
                self.id = aggregate_id


        class InMemoryEventStore[T]:
            def __init__(self) -> None: ...

            async def append_events(
                self, aggregate_id: object, events: list[object], expected_version: int
            ) -> None: ...
            async def get_events(self, aggregate_id: object) -> list[object]: ...
            async def get_current_version(self, aggregate_id: object) -> int: ...


        class MyAggregate(AggregateRoot[UUID, str]):
            def __init__(self, aggregate_id: UUID) -> None:
                super().__init__(aggregate_id)

            def _handle(self, event: Event[str]) -> None:
                pass


        event_store = InMemoryEventStore[str]()
        aggregate_id = UUID("00000000-0000-0000-0000-000000000001")
        repo = AggregateRepository[str, MyAggregate, UUID](
            event_store=event_store,
            aggregate_type=MyAggregate,
        )


        async def main() -> None:
            aggregate = MyAggregate(aggregate_id)
            await repo.save(aggregate)
            retrieved = await repo.get_by_id(aggregate_id)
        ```
    """

    _event_store: EventStoreBase[EventPayloadType]

    def __init__(
        self,
        event_store: EventStoreBase[EventPayloadType],
        aggregate_type: type[TAggregateRoot],
        storage: dict[TId, TAggregateRoot] | None = None,
    ) -> None:
        """Initialize the aggregate repository.

        Args:
            event_store: The event store for persisting domain events.
            aggregate_type: The aggregate root class. Used via
                its ``reconstitute`` classmethod when an aggregate
                must be rebuilt from stored events.
            storage: Optional in-memory storage for aggregate snapshots.

        """
        super().__init__(storage)
        self._event_store = event_store
        self._aggregate_type = aggregate_type

    async def save(self, aggregate: TAggregateRoot) -> None:
        """Save an aggregate and its state-changing events.

        Writes state-changing events to the event store first, then persists
        the aggregate snapshot. Publication-only events recorded through
        ``AggregateRoot.record_event`` remain available to the Unit of Work
        but are not stored in the aggregate event stream.

        The repository does not drain either event queue. The Unit of Work is
        the authoritative event-drain boundary after publication. Saving the
        same aggregate again before that drain attempts to append the same
        state changes and raises a concurrency error.

        The ``cast`` on ``state_changes`` bridges the gap between
        ``TAggregateRoot``'s bound (``AggregateRoot[UUID, Any]``) and the
        repository's ``EventPayloadType`` generic. The types are guaranteed to
        match at runtime by construction; the type system cannot express this
        cross-TypeVar-bound relationship (see PEP 695 and Pyright's
        ``reportGeneralTypeIssues``).

        Args:
            aggregate: The aggregate to save.

        Raises:
            EventStoreError: If the event store write fails (e.g., concurrency
                conflict, I/O error).

        """
        events = cast(list[Event[EventPayloadType]], aggregate.state_changes)
        aggregate_id: UUID | None = aggregate.id
        if events and aggregate_id is not None:
            version = aggregate.version.value - len(events)
            result = await self._event_store.append_events(
                aggregate_id, events, expected_version=version
            )
            if not result.is_ok:
                raise result.error
        await super().save(aggregate)

    async def get_by_id(self, entity_id: TId) -> TAggregateRoot | None:
        """Retrieve an aggregate by ID and replay its events.

        Checks the in-memory cache first; if not cached, replays the
        aggregate from the event store and caches the result so subsequent
        reads avoid a full replay.

        Args:
            entity_id: Unique identifier of the aggregate.

        Returns:
            The retrieved aggregate or None if not found.

        Raises:
            EventStoreError: If the event store read fails. Callers can
                distinguish infrastructure failures from "not found" (None).

        """
        aggregate = await super().get_by_id(entity_id)
        if aggregate is not None:
            return aggregate

        result = await self._event_store.get_events(cast(UUID, entity_id))

        if not result.is_ok:
            raise result.error

        events = result.value

        if not events:
            return None

        aggregate = self._aggregate_type.reconstitute(entity_id, events)
        await super().save(aggregate)

        return aggregate

__init__(event_store: EventStoreBase[EventPayloadType], aggregate_type: type[TAggregateRoot], storage: dict[TId, TAggregateRoot] | None = None) -> None

Initialize the aggregate repository.

Parameters:

Name Type Description Default
event_store EventStoreBase[EventPayloadType]

The event store for persisting domain events.

 required
aggregate_type type[TAggregateRoot]

The aggregate root class. Used via
its reconstitute classmethod when an aggregate
must be rebuilt from stored events.

 required
storage dict[TId, TAggregateRoot] | None

Optional in-memory storage for aggregate snapshots.

 None
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
def __init__(
    self,
    event_store: EventStoreBase[EventPayloadType],
    aggregate_type: type[TAggregateRoot],
    storage: dict[TId, TAggregateRoot] | None = None,
) -> None:
    """Initialize the aggregate repository.

    Args:
        event_store: The event store for persisting domain events.
        aggregate_type: The aggregate root class. Used via
            its ``reconstitute`` classmethod when an aggregate
            must be rebuilt from stored events.
        storage: Optional in-memory storage for aggregate snapshots.

    """
    super().__init__(storage)
    self._event_store = event_store
    self._aggregate_type = aggregate_type

save(aggregate: TAggregateRoot) -> None async

Save an aggregate and its state-changing events.

Writes state-changing events to the event store first, then persists
the aggregate snapshot. Publication-only events recorded through
AggregateRoot.record_event remain available to the Unit of Work
but are not stored in the aggregate event stream.

The repository does not drain either event queue. The Unit of Work is
the authoritative event-drain boundary after publication. Saving the
same aggregate again before that drain attempts to append the same
state changes and raises a concurrency error.

The cast on state_changes bridges the gap between
TAggregateRoot's bound (AggregateRoot[UUID, Any]) and the
repository's EventPayloadType generic. The types are guaranteed to
match at runtime by construction; the type system cannot express this
cross-TypeVar-bound relationship (see PEP 695 and Pyright's
reportGeneralTypeIssues).

Parameters:

Name Type Description Default
aggregate TAggregateRoot

The aggregate to save.

 required

Raises:

Type Description
EventStoreError

If the event store write fails (e.g., concurrency
conflict, I/O error).
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
async def save(self, aggregate: TAggregateRoot) -> None:
    """Save an aggregate and its state-changing events.

    Writes state-changing events to the event store first, then persists
    the aggregate snapshot. Publication-only events recorded through
    ``AggregateRoot.record_event`` remain available to the Unit of Work
    but are not stored in the aggregate event stream.

    The repository does not drain either event queue. The Unit of Work is
    the authoritative event-drain boundary after publication. Saving the
    same aggregate again before that drain attempts to append the same
    state changes and raises a concurrency error.

    The ``cast`` on ``state_changes`` bridges the gap between
    ``TAggregateRoot``'s bound (``AggregateRoot[UUID, Any]``) and the
    repository's ``EventPayloadType`` generic. The types are guaranteed to
    match at runtime by construction; the type system cannot express this
    cross-TypeVar-bound relationship (see PEP 695 and Pyright's
    ``reportGeneralTypeIssues``).

    Args:
        aggregate: The aggregate to save.

    Raises:
        EventStoreError: If the event store write fails (e.g., concurrency
            conflict, I/O error).

    """
    events = cast(list[Event[EventPayloadType]], aggregate.state_changes)
    aggregate_id: UUID | None = aggregate.id
    if events and aggregate_id is not None:
        version = aggregate.version.value - len(events)
        result = await self._event_store.append_events(
            aggregate_id, events, expected_version=version
        )
        if not result.is_ok:
            raise result.error
    await super().save(aggregate)

get_by_id(entity_id: TId) -> TAggregateRoot | None async

Retrieve an aggregate by ID and replay its events.

Checks the in-memory cache first; if not cached, replays the
aggregate from the event store and caches the result so subsequent
reads avoid a full replay.

Parameters:

Name Type Description Default
entity_id TId

Unique identifier of the aggregate.

 required

Returns:

Type Description
TAggregateRoot | None

The retrieved aggregate or None if not found.

Raises:

Type Description
EventStoreError

If the event store read fails. Callers can
distinguish infrastructure failures from "not found" (None).
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
async def get_by_id(self, entity_id: TId) -> TAggregateRoot | None:
    """Retrieve an aggregate by ID and replay its events.

    Checks the in-memory cache first; if not cached, replays the
    aggregate from the event store and caches the result so subsequent
    reads avoid a full replay.

    Args:
        entity_id: Unique identifier of the aggregate.

    Returns:
        The retrieved aggregate or None if not found.

    Raises:
        EventStoreError: If the event store read fails. Callers can
            distinguish infrastructure failures from "not found" (None).

    """
    aggregate = await super().get_by_id(entity_id)
    if aggregate is not None:
        return aggregate

    result = await self._event_store.get_events(cast(UUID, entity_id))

    if not result.is_ok:
        raise result.error

    events = result.value

    if not events:
        return None

    aggregate = self._aggregate_type.reconstitute(entity_id, events)
    await super().save(aggregate)

    return aggregate