Skip to content

Event Store Port

event_store_port

Event store port for event-sourced aggregates.

Defines the EventStorePort contract for appending and retrieving
domain events. The interface is agnostic of storage backend — in-memory,
relational, or event-native implementations are all supported.

EventStorePort

Bases: OutboundPort

Abstract base class for event stores that persist and retrieve domain events.

Responsibilities
  • Append domain events to an aggregate's event stream.
  • Retrieve events by version range from a stream.
  • Track and report the current version of each stream.
  • Enforce optimistic concurrency via expected_version.
Non-Responsibilities
  • Serialize or deserialize events — that belongs to infrastructure.
  • Validate event schemas or enforce event versioning.
  • Manage snapshots or compaction strategies.
Example 
store = MyEventStore[OrderEventData]()
await store.append_events(
    UUID("00000000-0000-0000-0000-000000000001"), [OrderPlaced(order_id="42")]
)
result = await store.get_events(
    UUID("00000000-0000-0000-0000-000000000001"), from_version=0
)
Source code in src/forging_blocks/application/ports/outbound/event_store_port.py 
 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
class EventStorePort[EventPayloadType](
    OutboundPort,
):
    """Abstract base class for event stores that persist and retrieve domain events.

    Responsibilities:
        - Append domain events to an aggregate's event stream.
        - Retrieve events by version range from a stream.
        - Track and report the current version of each stream.
        - Enforce optimistic concurrency via ``expected_version``.

    Non-Responsibilities:
        - Serialize or deserialize events — that belongs to infrastructure.
        - Validate event schemas or enforce event versioning.
        - Manage snapshots or compaction strategies.

    Example:
        ```python
        store = MyEventStore[OrderEventData]()
        await store.append_events(
            UUID("00000000-0000-0000-0000-000000000001"), [OrderPlaced(order_id="42")]
        )
        result = await store.get_events(
            UUID("00000000-0000-0000-0000-000000000001"), from_version=0
        )
        ```
    """

    @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/application/ports/outbound/event_store_port.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
@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/application/ports/outbound/event_store_port.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
@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/application/ports/outbound/event_store_port.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@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.

    """
    ...