Skip to content

🎛️ Event Sourcing

Building blocks

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

These are Aggregate and Repository base 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_sourcing import Aggregate, Repository
from event_sourcery.event_store import Event

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:

from event_sourcery.event_store import StreamUUID

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