Skip to content

In Memory Event Store

in_memory_event_store

In-memory implementation of the EventStorePort port.

This implementation stores events in memory and is suitable for testing
and development purposes.

InMemoryEventStore

Bases: EventStorePort

In-memory implementation of the EventStorePort port.

Stores events in a dictionary keyed by aggregate ID.

Source code in src/forging_blocks/infrastructure/in_memory_event_store.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
 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
class InMemoryEventStore(EventStorePort):
    """
    In-memory implementation of the EventStorePort port.

    Stores events in a dictionary keyed by aggregate ID.
    """

    def __init__(self) -> None:
        self._events: Dict[str, List[dict[str, Any]]] = defaultdict(list)
        self._snapshots: Dict[str, Dict[int, dict[str, Any]]] = defaultdict(dict)

    async def save_events(
        self,
        aggregate_id: str,
        events: List[dict[str, Any]],
        expected_version: Optional[int] = None,
    ) -> None:
        """
        Save a list of events for an aggregate.

        Args:
            aggregate_id: The unique identifier of the aggregate.
            events: The list of events to save.
            expected_version: The expected version of the aggregate.

        Raises:
            ConcurrencyError: If the expected version does not match the current version.
        """
        current_version = len(self._events[aggregate_id])

        if expected_version is not None and current_version != expected_version:
            raise ConcurrencyError(
                f"Concurrency conflict for aggregate {aggregate_id}: "
                f"expected version {expected_version}, but current version is {current_version}"
            )

        for event in events:
            event["version"] = current_version + 1
            event["aggregate_id"] = aggregate_id
            self._events[aggregate_id].append(event)
            current_version += 1

    async def get_events(
        self,
        aggregate_id: str,
        from_version: Optional[int] = None,
        to_version: Optional[int] = None,
    ) -> list[dict[str, object]]:
        """
        Retrieve events for an aggregate.

        Args:
            aggregate_id: The unique identifier of the aggregate.
            from_version: The starting version (inclusive).
            to_version: The ending version (inclusive).

        Returns:
            A list of events.
        """
        events = self._events.get(aggregate_id, [])

        if from_version is not None:
            events = [e for e in events if e.get("version", 0) >= from_version]

        if to_version is not None:
            events = [e for e in events if e.get("version", 0) <= to_version]

        return events

    async def get_snapshot(self, aggregate_id: str, version: int) -> dict[str, object] | None:
        """
        Retrieve a snapshot of an aggregate at a specific version.

        Args:
            aggregate_id: The unique identifier of the aggregate.
            version: The version of the snapshot.

        Returns:
            The snapshot data, or None if not found.
        """
        return self._snapshots.get(aggregate_id, {}).get(version)

    async def save_snapshot(
        self, aggregate_id: str, version: int, snapshot: dict[str, object]
    ) -> None:
        """
        Save a snapshot of an aggregate.

        Args:
            aggregate_id: The unique identifier of the aggregate.
            version: The version of the snapshot.
            snapshot: The snapshot data.
        """
        self._snapshots[aggregate_id][version] = snapshot

save_events(aggregate_id: str, events: List[dict[str, Any]], expected_version: Optional[int] = None) -> None async

Save a list of events for an aggregate.

Parameters:

Name Type Description Default
aggregate_id str

The unique identifier of the aggregate.

 required
events List[dict[str, Any]]

The list of events to save.

 required
expected_version Optional[int]

The expected version of the aggregate.

 None

Raises:

Type Description
ConcurrencyError

If the expected version does not match the current version.
Source code in src/forging_blocks/infrastructure/in_memory_event_store.py 
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
async def save_events(
    self,
    aggregate_id: str,
    events: List[dict[str, Any]],
    expected_version: Optional[int] = None,
) -> None:
    """
    Save a list of events for an aggregate.

    Args:
        aggregate_id: The unique identifier of the aggregate.
        events: The list of events to save.
        expected_version: The expected version of the aggregate.

    Raises:
        ConcurrencyError: If the expected version does not match the current version.
    """
    current_version = len(self._events[aggregate_id])

    if expected_version is not None and current_version != expected_version:
        raise ConcurrencyError(
            f"Concurrency conflict for aggregate {aggregate_id}: "
            f"expected version {expected_version}, but current version is {current_version}"
        )

    for event in events:
        event["version"] = current_version + 1
        event["aggregate_id"] = aggregate_id
        self._events[aggregate_id].append(event)
        current_version += 1

get_events(aggregate_id: str, from_version: Optional[int] = None, to_version: Optional[int] = None) -> list[dict[str, object]] async

Retrieve events for an aggregate.

Parameters:

Name Type Description Default
aggregate_id str

The unique identifier of the aggregate.

 required
from_version Optional[int]

The starting version (inclusive).

 None
to_version Optional[int]

The ending version (inclusive).

 None

Returns:

Type Description
list[dict[str, object]]

A list of events.
Source code in src/forging_blocks/infrastructure/in_memory_event_store.py 
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
async def get_events(
    self,
    aggregate_id: str,
    from_version: Optional[int] = None,
    to_version: Optional[int] = None,
) -> list[dict[str, object]]:
    """
    Retrieve events for an aggregate.

    Args:
        aggregate_id: The unique identifier of the aggregate.
        from_version: The starting version (inclusive).
        to_version: The ending version (inclusive).

    Returns:
        A list of events.
    """
    events = self._events.get(aggregate_id, [])

    if from_version is not None:
        events = [e for e in events if e.get("version", 0) >= from_version]

    if to_version is not None:
        events = [e for e in events if e.get("version", 0) <= to_version]

    return events

get_snapshot(aggregate_id: str, version: int) -> dict[str, object] | None async

Retrieve a snapshot of an aggregate at a specific version.

Parameters:

Name Type Description Default
aggregate_id str

The unique identifier of the aggregate.

 required
version int

The version of the snapshot.

 required

Returns:

Type Description
dict[str, object] | None

The snapshot data, or None if not found.
Source code in src/forging_blocks/infrastructure/in_memory_event_store.py 
83
84
85
86
87
88
89
90
91
92
93
94
async def get_snapshot(self, aggregate_id: str, version: int) -> dict[str, object] | None:
    """
    Retrieve a snapshot of an aggregate at a specific version.

    Args:
        aggregate_id: The unique identifier of the aggregate.
        version: The version of the snapshot.

    Returns:
        The snapshot data, or None if not found.
    """
    return self._snapshots.get(aggregate_id, {}).get(version)

save_snapshot(aggregate_id: str, version: int, snapshot: dict[str, object]) -> None async

Save a snapshot of an aggregate.

Parameters:

Name Type Description Default
aggregate_id str

The unique identifier of the aggregate.

 required
version int

The version of the snapshot.

 required
snapshot dict[str, object]

The snapshot data.

 required
Source code in src/forging_blocks/infrastructure/in_memory_event_store.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def save_snapshot(
    self, aggregate_id: str, version: int, snapshot: dict[str, object]
) -> None:
    """
    Save a snapshot of an aggregate.

    Args:
        aggregate_id: The unique identifier of the aggregate.
        version: The version of the snapshot.
        snapshot: The snapshot data.
    """
    self._snapshots[aggregate_id][version] = snapshot