Event Sourcing with UmaDB

This package supports using the Python eventsourcing library with UmaDB.

Installation

Add the Python eventsourcing package to your project with the umadb option. Run uv init to start a new project.

uv add "eventsourcing[umadb,pydantic]~=10.0.0a3"

Pydantic is recommended for modelling and serialising events.

After installing, you can start the UmaDB server with uv run umadb.

Getting started

Use the eventsourcing.dcb package to define a DCB application. Read the docs for more information.

from typing import TypedDict
from uuid import uuid4

from eventsourcing.domain import triggers
from eventsourcing.pydantic import DcbApplication, Decision, EnduringObject


class Dog(EnduringObject):
    class Registered(Decision):
        dog_id: str
        name: str

    class TrickAdded(Decision):
        dog_id: str
        trick: str

    @triggers(Registered)
    def __init__(self, *, name: str) -> None:
        self.name = name
        self.tricks: list[str] = []

    @triggers(TrickAdded)
    def add_trick(self, trick: str) -> None:
        self.tricks.append(trick)


class TrainingSchool(DcbApplication):
    def register(self, name: str) -> str:
        dog = Dog(name=name)
        self.repository.save(dog)
        return dog.id

    def add_trick(self, dog_id: str, trick: str) -> None:
        dog = self.repository.get(dog_id, Dog)
        dog.add_trick(trick)
        self.repository.save(dog)

    def get_dog(self, dog_id: str) -> DogSummary:
        dog = self.repository.get(dog_id, Dog)
        return {"name": dog.name, "tricks": tuple(dog.tricks)}


class DogSummary(TypedDict):
    name: str
    tricks: tuple[str, ...]

Configure the application to use UmaDB. Set environment variable PERSISTENCE_MODULE to 'eventsourcing_umadb', and set UMADB_URI to your UmaDB URI.

app = TrainingSchool(env={
    "PERSISTENCE_MODULE": "eventsourcing_umadb",
    "UMADB_URI": "http://127.0.0.1:50051",
})

The application's methods may be then called, from tests and user interfaces.

# Register dog.
dog_id = app.register("Fido")

# Add tricks.
app.add_trick(dog_id, "roll over")
app.add_trick(dog_id, "play dead")

# Get details.
dog = app.get_dog(dog_id)
assert dog["name"] == "Fido"
assert dog["tricks"] == ("roll over", "play dead")

For more information, please refer to the Python eventsourcing library and the UmaDB project.

Community

Join the Event Sourcing in Python Discord server today.

Contributors

Clone the GitHub repo and the use the following make commands.

Install Poetry.

make install-poetry

Install packages.

make install

Start UmaDB.

make start-umadb

Run tests.

make test

Stop UmaDB.

make stop-umadb

Check the formatting of the code.

make lint

Reformat the code.

make fmt

Tests belong in ./tests.

Edit package dependencies in pyproject.toml. Update installed packages (and the poetry.lock file) using the following command.

make update