eventsourcing

9.5.4 verified Fri May 01 auth: no python

A library for event sourcing in Python. It provides a framework for building applications using the event sourcing pattern, with support for aggregates, projections, and various persistence backends. Current version is 9.5.4, requiring Python >=3.10. Active development with regular releases.

pip install eventsourcing

Common errors

error ModuleNotFoundError: No module named 'eventsourcing.aggregate' ↓

cause Aggregate class moved to eventsourcing.domain in v9.

fix

Use 'from eventsourcing.domain import Aggregate'.

error TypeError: Can't instantiate abstract class Dog with abstract methods ↓

cause Domain aggregate class missing @event decorator on __init__.

fix

Add '@event('Registered')' above __init__.

error AttributeError: 'Dog' object has no attribute 'save' ↓

cause Calling save on the aggregate instance directly instead of through application.

fix

Use 'self.save(dog)' inside the Application method, not dog.save()

Warnings

breaking In v9, the domain model classes (Aggregate, Event, etc.) are no longer in the top-level package. They moved to eventsourcing.domain. Importing from old paths will raise ImportError. ↓

fix Change imports to from eventsourcing.domain import Aggregate, Event, etc.

breaking The 'event' decorator for methods is now required to mark event constructor methods. In v8, event methods were inferred; in v9 they must be explicitly decorated. ↓

fix Decorate __init__ and other event-applying methods with @event('EventName').

deprecated The 'sqlalchemy' persistence module has been deprecated in favor of 'sqlalchemy' as an optional dependency via eventsourcing[postgres] or eventsourcing[sqlite]. Direct import from eventsourcing.persistence.sqlalchemy may be removed. ↓

fix Use eventsourcing[postgres] or eventsourcing[sqlite] extras and configure via environment variables.

gotcha When using SQLite for persistence, the database file path must be set via environment variable DB_URI or passed to Application constructor. Not providing it will cause silent fallback to in-memory store which is not persisted. ↓

fix Set os.environ['DB_URI'] = 'sqlite:///path/to/db' before instantiating app.

Imports

Aggregate

wrong

from eventsourcing.aggregate import Aggregate

correct

from eventsourcing.domain import Aggregate

Aggregate moved to eventsourcing.domain in v9.

Event

wrong

from eventsourcing.event import Event

correct

from eventsourcing.domain import Event

Event classes are in eventsourcing.domain.

Application

wrong

from eventsourcing.app import Application

correct

from eventsourcing.application import Application

Application has been in eventsourcing.application since early versions.

LogEvent

wrong

from eventsourcing.events import LogEvent

correct

from eventsourcing.domain import LogEvent

LogEvent is part of the domain module.

Quickstart

Minimal example: define an aggregate with events, an application, and execute.

import os
from eventsourcing.application import Application
from eventsourcing.domain import Aggregate, event

class Dog(Aggregate):
    @event('Registered')
    def __init__(self, name):
        self.name = name
        self.tricks = []

    @event('TrickAdded')
    def add_trick(self, trick):
        self.tricks.append(trick)

class DogSchool(Application):
    def register_dog(self, name):
        dog = Dog(name)
        self.save(dog)
        return dog.id

    def add_trick(self, dog_id, trick):
        dog = self.repository.get(dog_id)
        dog.add_trick(trick)
        self.save(dog)

    def get_dog(self, dog_id):
        return self.repository.get(dog_id)

app = DogSchool()
dog_id = app.register_dog('Fido')
app.add_trick(dog_id, 'roll over')
dog = app.get_dog(dog_id)
print(f'{dog.name}: {dog.tricks}')