Event Store¶

The event store is the persistence layer for event sourcing. It handles appending new events to streams, reading them back, and managing stream lifecycle (including soft deletion). 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, event_types) — read events across all streams, optionally filtered by event types
• stream_exists(stream_id) — check whether a stream exists
• global_head_position() — return the highest global position across all streams, or -1 if empty
• read_positions(*, after_position, up_to_position) — return committed global positions in a range (used by gap detection)

IEventWriter — write-side operations:

• append_to_stream(stream_id, events, *, expected_version) — append events with optimistic concurrency
• delete_stream(stream_id) — soft-delete a stream (see Stream Deletion)

IEventStore combines both:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

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,
        event_types: Sequence[str] | None = None,
    ) -> list[StoredEvent]: ...

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

    async def global_head_position(self) -> int:
        """Return the highest global position, or -1 if empty."""
        ...

    async def read_positions(
        self,
        *,
        after_position: int,
        up_to_position: int,
    ) -> list[int]:
        """Return committed global positions in (after_position, up_to_position]."""
        ...


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

    async def delete_stream(self, stream_id: StreamId, /) -> None: ...


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

In-Memory Store¶

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

from waku.eventsourcing.store.in_memory import InMemoryEventStore

config = EventSourcingConfig(store=InMemoryEventStore)

PostgreSQL with SQLAlchemy¶

Prerequisites¶

Install the required extras:

uv add waku --extra eventsourcing --extra 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.

Stream Deletion¶

delete_stream() performs a soft delete — the stream is marked as deleted but its events are preserved for audit purposes.

await store.delete_stream(stream_id)

Calling delete_stream() on a nonexistent stream raises StreamNotFoundError. Calling it on an already-deleted stream is a no-op.

The SQLAlchemy store sets a deleted_at timestamp on the es_streams row and uses a JOIN filter to exclude deleted streams from read_all and read_positions queries. The in-memory store tracks deleted stream keys in a separate set.

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:

Database Schema Summary¶

All tables created by waku's event sourcing module:

from sqlalchemy import MetaData

from waku.eventsourcing.projection.lock.sqlalchemy import bind_lease_tables
from waku.eventsourcing.projection.sqlalchemy import bind_checkpoint_tables
from waku.eventsourcing.snapshot.sqlalchemy import bind_snapshot_tables
from waku.eventsourcing.store.sqlalchemy import bind_event_store_tables

metadata = MetaData()
es_tables = bind_event_store_tables(metadata)
bind_checkpoint_tables(metadata)
bind_lease_tables(metadata)
bind_snapshot_tables(metadata)

Further reading¶

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