Skip to content

Get started

Welcome! This page takes you from zero to storing and reading your first event in under a minute.

Installation

pip install python-event-sourcery

For production use, install the extra matching your storage of choice: sqlalchemy, kurrentdb or django, e.g.:

pip install "python-event-sourcery[sqlalchemy]"

Your first event

The example below uses the in-memory backend, so you can run it right away, without setting up a database.

Define an event by inheriting from the Event base class, then append it to a stream:

from event_sourcery import StreamId
from event_sourcery.backend import InMemoryBackend
from event_sourcery.event import Event

class InvoicePaid(Event):
    invoice_number: str

backend = InMemoryBackend()
event_store = backend.event_store

event_store.append(
    InvoicePaid(invoice_number="1003"),
    stream_id=StreamId(name="invoices/1003"),
)

Reading events

Events from a given stream can be read back at once:

recorded_events = event_store.load_stream(StreamId(name="invoices/1003"))
for recorded in recorded_events:
    print(recorded.event, recorded.version)

Output:

invoice_number='1003' 1

That's the essence of the library: events are defined as pydantic models, appended to named streams and read back together with their metadata.

What's next?

  • Use a real database — the in-memory backend is great for trying things out and for tests, but your app needs durable storage. See Integrate with your app.
  • React to events — process events asynchronously in another process with subscriptions and publish them reliably with the outbox.
  • Go full Event Sourcing — model your domain with aggregates and repositories. See the Event Sourcing recipe.