Skip to content

Event Store Port

event_store_port

Event store port for event-sourced aggregates.

Defines the EventStorePort protocol 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[object, object], Protocol

Protocol for event stores that persist and retrieve domain events.

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.
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
class EventStorePort[EventPayloadType](
    OutboundPort[object, object],
    Protocol,
):
    """Protocol for event stores that persist and retrieve domain events.

    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``.
    """

    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.
        """
        ...

    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.
        """
        ...

    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] 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 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
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] 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 
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
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] 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 
72
73
74
75
76
77
78
79
80
81
82
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.
    """
    ...