Event sourcing

Building blocks

Event Sourcery provides a few building blocks to work with event sourcing.

These are Aggregate, Repository and WrappedAggregate classes.

Usage

You start from defining your own aggregate inheriting from Aggregate.

There are three required attributes that need to be defined:

1. category class-level constant that will be added to all streams from all aggregates of this type
2. __init__ if defined, must not accept any arguments
3. __apply__ method that will change internal state of the aggregate based on the event applied during reading state from the event store

from event_sourcery.event import Event
from event_sourcery.event_sourcing import Aggregate, Repository

class SwitchedOn(Event):
    pass

class LightSwitch(Aggregate):
    category = "light_switch"  # 1

    def __init__(self) -> None:  # 2
        self._switched_on = False

    def __apply__(self, event: Event) -> None:  # 3
        match event:
            case SwitchedOn():
                self._switched_on = True
            case _:
                raise NotImplementedError(f"Unexpected event {type(event)}")

    def switch_on(self) -> None:
        if self._switched_on:
            return  # no op
        self._emit(SwitchedOn())

To work with aggregate, you need to create repository. You need an instance of EventStore to do so:

repository = Repository[LightSwitch](backend.event_store)

From now on, regardless if you want to work with a given aggregate for the first time or load existing one, you should use repository.aggregate context manager. It returns a WrappedAggregate — a wrapper that provides the aggregate instance along with stream metadata such as version, is_new, created_at, and updated_at:

from event_sourcery import StreamUUID

stream_id = StreamUUID(name="light_switch/1")
with repository.aggregate(stream_id, LightSwitch()) as wrapped:
    wrapped.aggregate.switch_on()
    wrapped.aggregate.switch_on()

The aggregate itself is accessed via wrapped.aggregate. The wrapper also exposes useful properties:

• wrapped.version — current version including pending (not yet persisted) changes
• wrapped.is_new — True if no events existed before this session
• wrapped.created_at / wrapped.updated_at — timestamps of first and last event in the stream
• wrapped.stream_id — the stream identity (UUID + category)