Event Store¶

The event store is the persistence layer for event sourcing. It handles appending new events to streams and reading them back. waku provides an in-memory implementation for development and a SQLAlchemy-based PostgreSQL adapter for production.

IEventStore Interface¶

The store interface is split into two protocols:

IEventReader — read-side operations:

• read_stream(stream_id, *, start, count) — read events from a single stream
• read_all(*, after_position, count) — read events across all streams by global position
• stream_exists(stream_id) — check whether a stream exists

IEventWriter — write-side operations:

• append_to_stream(stream_id, events, *, expected_version) — append events with optimistic concurrency

IEventStore combines both:

class IEventReader(abc.ABC):
    async def read_stream(
        self,
        stream_id: StreamId,
        /,
        *,
        start: int | StreamPosition = StreamPosition.START,
        count: int | None = None,
    ) -> list[StoredEvent]: ...

    async def read_all(
        self,
        *,
        after_position: int = -1,
        count: int | None = None,
    ) -> list[StoredEvent]: ...

    async def stream_exists(self, stream_id: StreamId, /) -> bool: ...


class IEventWriter(abc.ABC):
    async def append_to_stream(
        self,
        stream_id: StreamId,
        /,
        events: Sequence[EventEnvelope],
        *,
        expected_version: ExpectedVersion,
    ) -> int: ...


class IEventStore(IEventReader, IEventWriter, abc.ABC):
    pass

In-Memory Store¶

InMemoryEventStore is the default store used when no explicit store is configured:

config = EventSourcingConfig()  # uses InMemoryEventStore

It stores all events in memory with thread-safe locking via anyio.Lock. Suitable for development, testing, and prototyping.

PostgreSQL with SQLAlchemy¶

Prerequisites¶

Install the required extras:

uv add waku[eventsourcing,eventsourcing-sqla]

You also need a running PostgreSQL instance.

Step 1: Set Up Tables and Engine¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

import os
from collections.abc import AsyncIterator

from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine

from waku.eventsourcing.store.sqlalchemy.tables import bind_event_store_tables

DATABASE_URL = os.environ['DATABASE_URL']

metadata = MetaData()
tables = bind_event_store_tables(metadata)
engine = create_async_engine(DATABASE_URL, echo=False)


async def create_session(engine_: AsyncEngine) -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine_, expire_on_commit=False) as session:
        yield session

bind_event_store_tables() binds the event store table definitions to your metadata instance.

Step 2: Configure Event Sourcing¶

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from waku.eventsourcing import EventSourcingConfig
from waku.eventsourcing.serialization.json import JsonEventSerializer
from waku.eventsourcing.store.sqlalchemy.store import make_sqlalchemy_event_store

from app.database import tables

es_config = EventSourcingConfig(
    store=make_sqlalchemy_event_store(tables),
    event_serializer=JsonEventSerializer,
)

make_sqlalchemy_event_store() is a factory that binds the tables to SqlAlchemyEventStore. It returns a callable that Dishka uses to construct the store with its remaining dependencies (session, serializer, registry, upcaster chain) injected automatically.

Step 3: Create Tables¶

Create the tables during application startup — typically in a lifespan handler or migration script:

async with engine.begin() as conn:
    await conn.run_sync(metadata.create_all)

This creates two tables:

es_streams¶

es_events¶

Unique constraints:

• uq_es_events_stream_id_position — (stream_id, position)
• uq_es_events_idempotency_key — (stream_id, idempotency_key)

Indexes:

• ix_es_events_global_position — on global_position (used by catch-up projections)
• ix_es_events_event_type — on event_type (used for type-filtered reads)

# alembic/env.py
from waku.eventsourcing.store.sqlalchemy import bind_event_store_tables

target_metadata = MetaData()
bind_event_store_tables(target_metadata)

Idempotency¶

Network retries and client resubmissions can cause duplicate event appends. waku prevents this through idempotency keys — client-provided deduplication tokens attached to each EventEnvelope.

EventEnvelope¶

EventEnvelope wraps a domain event with an idempotency_key for deduplication:

@dataclass(frozen=True, slots=True, kw_only=True)
class EventEnvelope:
    domain_event: INotification
    idempotency_key: str
    metadata: EventMetadata = field(default_factory=EventMetadata)

The idempotency_key is required and must be non-empty. It is scoped per stream — the same key can exist in different streams without conflict.

Deduplication Semantics¶

When appending events, the store checks idempotency keys with all-or-nothing semantics:

This means a full retry of the same batch is safe (idempotent), but mixing old and new events in a single append is rejected as an inconsistency.

Exceptions¶

Both exceptions are in waku.eventsourcing.exceptions:

Storage¶

The SQLAlchemy store persists idempotency keys in a dedicated column on es_events with a composite unique constraint (stream_id, idempotency_key). The in-memory store tracks keys per stream in a dictionary with an async lock for thread safety.

Metadata Enrichment¶

IMetadataEnricher allows you to add contextual metadata (correlation IDs, user identity, trace context) to every event before it is persisted.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

from waku.eventsourcing import EventMetadata, IMetadataEnricher


class CorrelationIdEnricher(IMetadataEnricher):
    def __init__(self, correlation_id: str) -> None:
        self._correlation_id = correlation_id

    def enrich(self, metadata: EventMetadata, /) -> EventMetadata:
        return EventMetadata(
            correlation_id=self._correlation_id,
            causation_id=metadata.causation_id,
            extra=metadata.extra,
        )

Register enrichers in the config:

es_config = EventSourcingConfig(
    enrichers=[CorrelationIdEnricher],
)

Enrichers are registered as DI providers, so Dishka resolves their constructor dependencies at runtime. The store calls each enricher's enrich() method in order before writing events.

EventMetadata¶

EventMetadata is a frozen dataclass attached to every event envelope:

StoredEvent¶

StoredEvent is the fully-hydrated event record returned when reading from the store:

Further reading¶

• Projections — build read models from event streams
• Schema Evolution — upcasting and event versioning
• Snapshots — optimize loading for long-lived aggregates
• Testing — in-memory event store for integration tests