Skip to content

Event Store Base

event_store_base

Event store base class compliant with the application EventStorePort contract.

Defines the EventStoreBase abstract interface for appending and retrieving
domain events for aggregate roots following the Event Sourcing pattern.

EventStoreBase

Bases: ABC

Base class for event stores.

Implementations must handle
  • Appending new events to an event stream.
  • Retrieving events within an optional version range.
  • Tracking the current version of each event stream.
  • Optimistic concurrency via expected_version.
Example 
class AccountDebited:
    def __init__(self, amount: int) -> None:
        self.amount = amount


class InMemoryEventStore:
    async def append_events(self, aggregate_id, events): ...
    async def get_events(self, aggregate_id): ...


store = InMemoryEventStore()
await store.append_events(
    UUID("00000000-0000-0000-0000-000000000001"), [AccountDebited(amount=50)]
)
events = await store.get_events(UUID("00000000-0000-0000-0000-000000000001"))
Source code in src/forging_blocks/infrastructure/event_stores/event_store_base.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
class EventStoreBase[EventPayloadType](ABC):
    """Base class for event stores.

    Implementations must handle:
      - Appending new events to an event stream.
      - Retrieving events within an optional version range.
      - Tracking the current version of each event stream.
      - Optimistic concurrency via ``expected_version``.

    Example:
        ```python
        class AccountDebited:
            def __init__(self, amount: int) -> None:
                self.amount = amount


        class InMemoryEventStore:
            async def append_events(self, aggregate_id, events): ...
            async def get_events(self, aggregate_id): ...


        store = InMemoryEventStore()
        await store.append_events(
            UUID("00000000-0000-0000-0000-000000000001"), [AccountDebited(amount=50)]
        )
        events = await store.get_events(UUID("00000000-0000-0000-0000-000000000001"))
        ```
    """

    @abstractmethod
    async def append_events(
        self,
        aggregate_id: UUID,
        events: Sequence[Event[EventPayloadType]],
        expected_version: int | None = None,
    ) -> Result[int, EventStoreError]:
        """Append events to an aggregate's event stream.

        Args:
            aggregate_id: The aggregate identifier.
            events: The domain events to append.
            expected_version: Expected current version for optimistic
                concurrency. ``None`` means skip the check.

        Returns:
            A ``Result`` containing the new stream version on success or an
            ``EventStoreError`` on failure.

        """

    @abstractmethod
    async def get_events(
        self,
        aggregate_id: UUID,
        from_version: int | None = None,
        to_version: int | None = None,
    ) -> Result[Sequence[Event[EventPayloadType]], EventStoreError]:
        """Retrieve events from an aggregate's event stream.

        Args:
            aggregate_id: The aggregate identifier.
            from_version: Start version (inclusive). ``None`` means from
                the beginning.
            to_version: End version (inclusive). ``None`` means until
                the end.

        Returns:
            A ``Result`` containing the events on success or an
            ``EventStoreError`` on failure.

        """

    @abstractmethod
    async def get_current_version(self, aggregate_id: UUID) -> Result[int, EventStoreError]:
        """Retrieve the current version of an aggregate's event stream.

        Args:
            aggregate_id: The aggregate identifier.

        Returns:
            A ``Result`` containing the latest version number (0 for empty
            streams) on success, or an ``EventStoreError`` on failure.

        """

append_events(aggregate_id: UUID, events: Sequence[Event[EventPayloadType]], expected_version: int | None = None) -> Result[int, EventStoreError] abstractmethod async

Append events to an aggregate's event stream.

Parameters:

Name Type Description Default
aggregate_id UUID

The aggregate identifier.

 required
events Sequence[Event[EventPayloadType]]

The domain events to append.

 required
expected_version int | None

Expected current version for optimistic
concurrency. None means skip the check.

 None

Returns:

Type Description
Result[int, EventStoreError]

A Result containing the new stream version on success or an
Result[int, EventStoreError]

EventStoreError on failure.
Source code in src/forging_blocks/infrastructure/event_stores/event_store_base.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@abstractmethod
async def append_events(
    self,
    aggregate_id: UUID,
    events: Sequence[Event[EventPayloadType]],
    expected_version: int | None = None,
) -> Result[int, EventStoreError]:
    """Append events to an aggregate's event stream.

    Args:
        aggregate_id: The aggregate identifier.
        events: The domain events to append.
        expected_version: Expected current version for optimistic
            concurrency. ``None`` means skip the check.

    Returns:
        A ``Result`` containing the new stream version on success or an
        ``EventStoreError`` on failure.

    """

get_events(aggregate_id: UUID, from_version: int | None = None, to_version: int | None = None) -> Result[Sequence[Event[EventPayloadType]], EventStoreError] abstractmethod async

Retrieve events from an aggregate's event stream.

Parameters:

Name Type Description Default
aggregate_id UUID

The aggregate identifier.

 required
from_version int | None

Start version (inclusive). None means from
the beginning.

 None
to_version int | None

End version (inclusive). None means until
the end.

 None

Returns:

Type Description
Result[Sequence[Event[EventPayloadType]], EventStoreError]

A Result containing the events on success or an
Result[Sequence[Event[EventPayloadType]], EventStoreError]

EventStoreError on failure.
Source code in src/forging_blocks/infrastructure/event_stores/event_store_base.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
@abstractmethod
async def get_events(
    self,
    aggregate_id: UUID,
    from_version: int | None = None,
    to_version: int | None = None,
) -> Result[Sequence[Event[EventPayloadType]], EventStoreError]:
    """Retrieve events from an aggregate's event stream.

    Args:
        aggregate_id: The aggregate identifier.
        from_version: Start version (inclusive). ``None`` means from
            the beginning.
        to_version: End version (inclusive). ``None`` means until
            the end.

    Returns:
        A ``Result`` containing the events on success or an
        ``EventStoreError`` on failure.

    """

get_current_version(aggregate_id: UUID) -> Result[int, EventStoreError] abstractmethod async

Retrieve the current version of an aggregate's event stream.

Parameters:

Name Type Description Default
aggregate_id UUID

The aggregate identifier.

 required

Returns:

Type Description
Result[int, EventStoreError]

A Result containing the latest version number (0 for empty
Result[int, EventStoreError]

streams) on success, or an EventStoreError on failure.
Source code in src/forging_blocks/infrastructure/event_stores/event_store_base.py 
88
89
90
91
92
93
94
95
96
97
98
99
@abstractmethod
async def get_current_version(self, aggregate_id: UUID) -> Result[int, EventStoreError]:
    """Retrieve the current version of an aggregate's event stream.

    Args:
        aggregate_id: The aggregate identifier.

    Returns:
        A ``Result`` containing the latest version number (0 for empty
        streams) on success, or an ``EventStoreError`` on failure.

    """