Skip to content

Aggregate Repository

aggregate_repository

Aggregate RepositoryPort implementation.

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

AggregateRepository

Bases: BaseRepository[TAggregateRoot, TId]

RepositoryPort for AggregateRoot persistence with event sourcing.

Extends BaseRepository with event store integration for
event-sourced aggregates.

Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
14
15
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
class AggregateRepository[TAggregateRoot: AggregateRoot[Any, Any], TId: Any](
    BaseRepository[TAggregateRoot, TId]
):
    """RepositoryPort for AggregateRoot persistence with event sourcing.

    Extends BaseRepository with event store integration for
    event-sourced aggregates.
    """

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

        Args:
            event_store: The event store for persisting domain events.
            storage: Optional in-memory storage for aggregate snapshots.
        """
        super().__init__(storage)
        self._event_store = event_store

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

        Args:
            aggregate: The aggregate to save.
        """
        # Save the aggregate snapshot
        await super().save(aggregate)

        # Persist uncommitted events to the event store
        events = aggregate.collect_events()
        if events:
            await self._event_store.save_events(
                aggregate_id=str(aggregate.id),
                events=[event.to_dict() for event in events],
                expected_version=aggregate.version.value - len(events),
            )

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

        Args:
            id: Unique identifier of the aggregate.

        Returns:
            The retrieved aggregate or None if not found.
        """
        # First check in-memory storage
        aggregate = await super().get_by_id(id)
        if aggregate is not None:
            return aggregate

        # If not in memory, try to load from event store
        events = await self._event_store.get_events(str(id))
        if not events:
            return None

        # Reconstruct aggregate from events
        # This requires the aggregate type to have a factory method
        # For now, return None - concrete implementations should override
        return None

__init__(event_store: EventStorePort, storage: dict[TId, TAggregateRoot] | None = None) -> None

Initialize the aggregate repository.

Parameters:

Name Type Description Default
event_store EventStorePort

The event store for persisting domain 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 
23
24
25
26
27
28
29
30
31
32
33
34
35
def __init__(
    self,
    event_store: EventStorePort,
    storage: dict[TId, TAggregateRoot] | None = None,
) -> None:
    """Initialize the aggregate repository.

    Args:
        event_store: The event store for persisting domain events.
        storage: Optional in-memory storage for aggregate snapshots.
    """
    super().__init__(storage)
    self._event_store = event_store

save(aggregate: TAggregateRoot) -> None async

Save an aggregate and its uncommitted events.

Parameters:

Name Type Description Default
aggregate TAggregateRoot

The aggregate to save.

 required
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
async def save(self, aggregate: TAggregateRoot) -> None:
    """Save an aggregate and its uncommitted events.

    Args:
        aggregate: The aggregate to save.
    """
    # Save the aggregate snapshot
    await super().save(aggregate)

    # Persist uncommitted events to the event store
    events = aggregate.collect_events()
    if events:
        await self._event_store.save_events(
            aggregate_id=str(aggregate.id),
            events=[event.to_dict() for event in events],
            expected_version=aggregate.version.value - len(events),
        )

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

Retrieve an aggregate by ID and replay its events.

Parameters:

Name Type Description Default
id TId

Unique identifier of the aggregate.

 required

Returns:

Type Description
TAggregateRoot | None

The retrieved aggregate or None if not found.
Source code in src/forging_blocks/infrastructure/repositories/aggregate_repository.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
async def get_by_id(self, id: TId) -> TAggregateRoot | None:  # noqa: A002
    """Retrieve an aggregate by ID and replay its events.

    Args:
        id: Unique identifier of the aggregate.

    Returns:
        The retrieved aggregate or None if not found.
    """
    # First check in-memory storage
    aggregate = await super().get_by_id(id)
    if aggregate is not None:
        return aggregate

    # If not in memory, try to load from event store
    events = await self._event_store.get_events(str(id))
    if not events:
        return None

    # Reconstruct aggregate from events
    # This requires the aggregate type to have a factory method
    # For now, return None - concrete implementations should override
    return None