Event Store

event_store

Event Store infrastructure.

This module provides the EventStorePort port and its in-memory implementation.

EventStorePort

Bases: ABC

EventStorePort port.

The EventStorePort is responsible for persisting and retrieving events
for aggregate roots. It follows the Event Sourcing pattern.

Source code in src/forging_blocks/infrastructure/event_store.py

class EventStorePort(ABC):
    """
    EventStorePort port.

    The EventStorePort is responsible for persisting and retrieving events
    for aggregate roots. It follows the Event Sourcing pattern.
    """

    @abstractmethod
    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.
        """
        pass

    @abstractmethod
    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.
        """
        pass

    @abstractmethod
    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.
        """
        pass

    @abstractmethod
    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.
        """
        pass

save_events(aggregate_id: str, events: List[dict[str, Any]], expected_version: Optional[int] = None) -> None abstractmethod 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/event_store.py

@abstractmethod
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.
    """
    pass

get_events(aggregate_id: str, from_version: Optional[int] = None, to_version: Optional[int] = None) -> list[dict[str, object]] abstractmethod 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/event_store.py

@abstractmethod
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.
    """
    pass

get_snapshot(aggregate_id: str, version: int) -> dict[str, object] | None abstractmethod 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/event_store.py

@abstractmethod
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.
    """
    pass

save_snapshot(aggregate_id: str, version: int, snapshot: dict[str, object]) -> None abstractmethod 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/event_store.py

@abstractmethod
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.
    """
    pass

ConcurrencyError

Bases: Error[dict[str, object]]

Raised when a concurrency conflict is detected.

Source code in src/forging_blocks/infrastructure/event_store.py

class ConcurrencyError(Error[dict[str, object]]):
    """
    Raised when a concurrency conflict is detected.
    """

    def __init__(self, message: str) -> None:
        super().__init__(ErrorMessage(message))