Guide

Requirements

Package	Required	Purpose
quadkit	Yes	Core framework
quadkit-contracts	Yes	Protocol definitions
a queue extension	Optional	Event bus backends
a resilience extension	Optional	Retry policies

Problem

Building event-driven systems is hard. Without a framework you end up writing the same plumbing — event storage, handler registration, bus dispatching, projection management — for every service. quadkit-events gives you a standard, DI-friendly foundation for CQRS and Event Sourcing so you can focus on domain logic.

Mental Model

quadkit-events is built around three message types that flow through three buses:

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│   Command   │────▶│ CommandBus   │────▶│  Handler    │
│  (mutate)   │     │  dispatch()  │     │  (domain)   │
└─────────────┘     └──────────────┘     └──────┬──────┘
                                                │ emits
                                                ▼
┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│  DomainEvent│────▶│  EventBus    │────▶│  Handler    │
│  (fact)     │     │  publish()   │     │  (side fx)  │
└─────────────┘     └──────────────┘     └─────────────┘

┌─────────────┐     ┌──────────────┐     ┌─────────────┐
│    Query    │────▶│  QueryBus    │────▶│  Handler    │
│  (read)     │     │  execute()   │     │  (project)  │
└─────────────┘     └──────────────┘     └─────────────┘

• Commands represent intent to change state. They are named imperatively (CreateOrder, CancelInvoice). Exactly one handler per command type.
• Events are facts that have already happened (OrderCreated, PaymentReceived). Zero or more handlers subscribe to each event type.
• Queries request data. Like commands, they route to exactly one handler.

Core Concepts

Domain Events

Domain events are immutable facts. They extend DomainEvent from quadkit.contracts.domain:

from quadkit.contracts.domain import DomainEvent


class OrderShipped(DomainEvent):
    order_id: str
    shipped_at: str
    tracking_number: str

Events are published to the EventBusProtocol, which dispatches them to all registered subscribers.

Commands and the Command Bus

Commands represent an intent to mutate state. They extend the Command base class from quadkit.contracts.events:

from quadkit.events.messages.command import Command


class SubmitOrder(Command):
    order_id: str
    items: list[dict]

Dispatch a command via the CommandBusProtocol:

command_bus = await container.resolve(CommandBusProtocol)
result = await command_bus.dispatch(SubmitOrder(order_id="ord-1", items=[]))

Handlers are registered using the @command_handler decorator:

from quadkit.events import command_handler


@command_handler
class SubmitOrderHandler:
    async def handle(self, command: SubmitOrder) -> None:
        # domain logic here
        ...

Event Bus and Handlers

The EventBusProtocol publishes domain events to multiple handlers. Subscribe with @event_handler:

from quadkit.events import event_handler
from quadkit.contracts.events import EventHandlerProtocol


@event_handler
class OrderEmailHandler(EventHandlerProtocol):
    async def handle(self, event: OrderShipped) -> Result[None, EventError]:
        await email_service.send_shipping_notification(event.order_id)
        return Ok(None)

Note

Event handlers return Result[None, EventError]. Failures are logged and optionally routed to a dead-letter queue based on EventBusConfig.

Event Store

The event store persists events to a durable backend. quadkit-events supports:

Backend	Config	Extra
In-Memory	InMemoryEventStoreConfig	(built-in)
PostgreSQL	PostgresEventStoreConfig	quadkit-events[postgres]
MongoDB	MongoDBEventStoreConfig	quadkit-events[mongo]
SQLite	SqliteConfig	quadkit-events[sqlite]

Configure via EventsConfig.event_store_backend:

from quadkit.events.config import EventsConfig
from quadkit.events.types import EventStoreBackend

config = EventsConfig(
    event_store_backend=EventStoreBackend.POSTGRES,
    postgres=PostgresEventStoreConfig(
        dsn=SecretStr("postgresql://localhost/events"),
    ),
)

Aggregates

Aggregates group domain logic and enforce invariants. They extend AggregateRoot:

from quadkit.events.aggregates.aggregate import AggregateRoot
from quadkit.events import AggregateStatus


class Order(AggregateRoot):
    def __init__(self, order_id: str) -> None:
        super().__init__(order_id)
        self._status = AggregateStatus.ACTIVE
        self._items: list[str] = []

    def add_item(self, item_id: str) -> None:
        self._items.append(item_id)
        self.record_event(ItemAdded(order_id=self.id, item_id=item_id))

Use EventSourcingRepository to load and save aggregates:

from quadkit.events import EventSourcingRepository


class OrderRepository(EventSourcingRepository):
    pass

Projections

Projections build read models from event streams:

from quadkit.contracts.events import ProjectionProtocol


class OrderSummaryProjection(ProjectionProtocol):
    def apply(self, event: OrderCreated) -> None:
        db.insert("order_summary", {
            "order_id": event.order_id,
            "customer": event.customer,
        })

Sagas

Sagas orchestrate long-running business processes across multiple aggregates:

from quadkit.contracts.workflow import SagaProtocol


class OrderFulfillmentSaga(SagaProtocol):
    async def handle(self, event: OrderCreated) -> None:
        await self.dispatch(ReserveInventory(order_id=event.order_id))

Typical Usage

import asyncio
from dataclasses import dataclass

from quadkit import Application
from quadkit.contracts.domain import DomainEvent
from quadkit.di.module import module, Module
from quadkit.events import (
    EventBusProtocol,
    EventsModule,
    command_handler,
    event_handler,
)
from quadkit.events.config import EventsConfig
from quadkit.events.messages.command import Command


# Events
class OrderCreated(DomainEvent):
    order_id: str


# Commands
@dataclass
class CreateOrder(Command):
    order_id: str


@command_handler
class CreateOrderHandler:
    async def handle(self, command: CreateOrder) -> None:
        print(f"Creating order {command.order_id}")
        # publish event via injected EventBusProtocol


@event_handler
class OrderNotificationHandler:
    async def handle(self, event: OrderCreated) -> None:
        print(f"Notifying about order {event.order_id}")


@module(imports=[EventsModule.configure()])
class AppModule(Module):
    pass


async def main() -> None:
    async with Application.boot(name="app", modules=[AppModule]) as app:
        bus = await app.container.resolve(EventBusProtocol)
        event = OrderCreated(order_id="ord-1")
        result = await bus.publish(event)
        if result.is_ok():
            print("Event published")


asyncio.run(main())

Local SQLite event streaming

In the local SqliteEventStore, stream_all(from_position=...) and stream_by_type(..., from_position=...) yield events after the supplied SQLite global sequence, with that sequence in event.sequence_number. The store stamps a copy after upcasting; custom serializers must return copyable objects that can hold the position. A failure to attach it raises rather than silently yielding an uncheckpointable event.

For stream_all(partition=p, total_partitions=n), both values are required, n must be positive, and 0 <= p < n. The local implementation scans ordered batches and assigns each stream using the first eight hex digits of the MD5 of its UTF-8 stream ID modulo n, matching the current memory and Redis rules. This is a non-cryptographic partition assignment. It does not require a SQLite hash extension, but each partition may scan rows belonging to other partitions; this can be expensive. Changing the partition count changes assignment. Older custom SQLite STR_HASH implementations and other backends may use a different rule, so migrating consumers or checkpoints requires an explicit migration plan.

These positions alone do not provide reliable delivery. The local handler-driven EventSubscriber now stops a scan at a handler failure and retries from the last successfully handled position on its next pass; later events wait behind a failing event. A handler may run more than once, so side effects must be idempotent. Missing, non-integer or non-advancing stream cursors stop the subscription in ERROR rather than being dispatched; this can affect backends that do not expose their global cursor. A checkpoint-store exception also stops the consumer. The separate events() iterator advances its position before yielding and is not an acknowledgement or retry mechanism.

The tests use an in-memory checkpoint store, not durable or concurrent group processing. Durable checkpoint storage, lease expiry/renewal, multi-process safety, dead-letter policy and recovery require separate validation. Snapshot correctness and histories lost before SQLite migration remain separate concerns. The generated event-effect path is not enabled by these local streaming changes.

In-memory event store limitations

The local in-memory store stages whole append batches and supports declared frozen/slotted dataclass events as well as legacy dynamic events. It shallowly copies each event, so nested mutable payload objects can still be shared with callers; returned events are not isolated copies either. Treat event payloads as immutable.

The local store tracks an absolute per-stream version across retention and compaction, so stale expected versions are rejected. A read that requests an evicted or compacted prefix raises EventStoreError instead of returning an incomplete history; callers must use a suitable snapshot and request only retained versions. This stricter behavior can affect existing callers.

These high-water marks exist only for the lifetime of the process and are not a durability or concurrency guarantee across processes. The global log still retains events evicted from individual streams, so the per-stream limit is not a total memory bound. Snapshot correctness and version continuity in other backends remain separate concerns.

Stream existence and information after compaction

For the local in-memory and SQLite stores, stream_exists() uses the logical stream version rather than loading retained events. It remains true after full compaction until the stream is explicitly reset (or the in-memory store is cleared). get_stream_info() reports the logical high-water version and the number of currently retained events as event_count. Its aggregate_type and updated_at describe retained events; created_at is unknown once the original prefix is gone. After full compaction, these metadata fields are None. The outbox and streaming wrappers delegate these helpers to their inner store.

EventSourcingRepository.exists() tests for a positive reported stream version; this is not proof that replay succeeds. It does not count a snapshot-only, zero-version aggregate, even though get() can load such a snapshot. Existence checks now require the async version capability. They trust the backend’s reported version, so other stores’ failure and compaction behavior must be validated separately.

SQLite compaction and migration

The local SQLite event store creates a companion table named <events_table>_stream_state and reserves that name for its compacted-prefix markers. It records the prefix and deletes events in the same transaction. Stream versions continue after partial or full compaction, including after a reopen. Reads that request a missing prefix, or encounter an internal version gap, raise EventStoreError; a covering snapshot can be used to request only the retained suffix. Fully compacted streams remain in get_stream_ids(). delete_stream() explicitly deletes both events and the marker, resetting that stream’s version. These stricter semantics may require caller changes.

With auto_create_tables=False, provision the companion table with the event schema before connecting (the event store’s get_create_events_table_sql() includes its DDL); a missing or incompatible column schema fails connection. Reserve the derived table name and do not reuse it for unrelated data. Schema checks are not a comprehensive migration or ownership verifier. Back up and audit historical data before migration: a legacy retained suffix reveals a missing prefix, but a stream fully compacted before this upgrade has no rows from which to recover its erased version. No automatic recovery is possible. Stop older writers and compactors before migrating; a row at or below a marker is rejected, but mixed-version operation cannot generally be detected or made safe. This is not a multi-process concurrency proof. Bounds checking scans a stream’s retained rows and can add cost for long streams; benchmark your load.

SQLite event payloads

Register each declared dataclass event class with store.register_event_type when constructing a SQLite store, including after restart. For registered DomainEvent dataclasses, the local serializer restores declared types and checks that persisted JSON values survive reconstruction; missing fields may receive declared defaults, but unknown or lossy values raise EventStoreError. This stricter check may reject older payloads whose representation changes on reconstruction, so test historical data and plan migrations before upgrading.

For default-serialized domain events, SQLite now checks that a payload’s event_id agrees with the row’s UUID on read, type search and streaming. If a legacy payload omitted event_id, the row supplies it; invalid or conflicting IDs raise instead of returning a different identity. New domain-event appends also reject conflicting id/event_id values before committing. Audit old rows before upgrading: existing conflicting IDs now fail to load.

Plain legacy event subclasses retain dynamic attributes and restore known base identity/time fields, but arbitrary custom attributes are not type-restored. Unregistered types fall back to generic DomainEvent, losing their original class and typed custom fields. Custom serializers and registered non-domain classes retain their own identity and deserialization contract. These checks are not authentication or complete row integrity: custom deserializers and upcasters are trusted, and stream/aggregate IDs, versions, timestamps, and payload event_type are not cross-checked. SQLite indexes the Python class name in its type column, which may differ from an explicitly supplied payload event_type. These rules do not prove aggregate replay, schema migration or outbox transport safety.

SQLite snapshots

The local SQLite snapshot store checks provider results, wraps saves and cleanup in transactions, and verifies the retained count after cleanup. keep_count=0 deletes all snapshots for that aggregate; negative or non-integer values are rejected. Duplicate aggregate/version saves are not replacements. The current schema has no snapshot metadata column, so saving non-empty metadata now raises instead of silently discarding it. Snapshot state is serialized as JSON and is not guaranteed to preserve arbitrary Python field types.

These checks were exercised on local direct and pooled SQLite providers; the cleanup verification checks counts, not the identity of every retained row. They do not guarantee arbitrary provider transaction behavior, durable schema migration, or correct aggregate replay. In particular, snapshot correctness and all forms of persisted history loss are not verified.

Snapshot manager and aggregate replay

SnapshotManager.force_snapshot() uses the store’s logical stream version rather than the number of retained events. The store must expose an async get_stream_version() capability; stores without it now fail closed. Pass expected_version to check the version associated with the supplied state. The manager also rejects a version change observed during state extraction. The caller still owns the correctness of the extracted state and must coordinate concurrent writers: the version checks and snapshot save are not atomic, and other backends’ version behavior has not been established by these tests.

EventSourcingRepository.force_snapshot() rejects pending events and checks that the persisted stream version matches the aggregate version before saving. Direct create_snapshot() and custom state extractors remain caller-trusted. After a normal append, save_and_maybe_snapshot() requires the acknowledged version to equal the expected version plus the event count before creating a snapshot. A bad acknowledgement may nevertheless follow a committed append; reconcile the stream before retrying. cleanup_old_snapshots(keep_count=0) now explicitly removes all snapshots for that aggregate; None uses the configured default. Negative or non-integer counts are rejected.

Local SnapshotManager.load_with_snapshot() and EventSourcingRepository.get() require an asynchronous get_stream_version() capability, even for streams without a snapshot. They compare the logical version before and after reading, reject a snapshot ahead of that version, and require one returned event per version in the requested suffix. A concurrent version change causes a failure rather than a partial aggregate; callers may retry. Custom stores that implement only the smaller EventStoreProtocol, or that paginate read implicitly, must provide compatible version and complete-read behavior. This stricter contract adds two version queries per load and may affect existing integrations.

The repository also checks snapshot identity/type and state, restores the requested identity and snapshot version, and raises EventStoreError rather than returning an aggregate after a failed restore or replay. Unknown/private snapshot fields, conflicting snapshot identity/version, and events with a different aggregate ID are rejected. For the standard AggregateRoot._apply, every replayed event must have a matching handler; previously unhandled events were silently ignored. Custom _apply or history implementations remain responsible for their own event handling.

These count and version checks trust the store’s reported high-water mark, ordering and event contents. They cannot detect same-count substitutions or reordering, an ABA delete/recreate, silent failures that report a consistent empty stream, or versions erased by legacy full compaction. They do not prove that a snapshot represents its history or that JSON-decoded values have the Python types expected by an aggregate. The version checks and read are not a single transactional view, and snapshot, event and event-bus side effects are not atomic; a failure after an append may require reconciliation. Validate existing snapshots, event registrations and backend behavior before upgrading.

Best Practices

• Keep events small. Each event represents one fact. Don’t bundle unrelated state changes.
• Make handlers idempotent. Events may be delivered more than once. Use idempotency keys or upsert semantics.
• Version your schemas. Use schema evolution tools for long-lived event stores.
• Use the outbox pattern for reliable multi-service event publishing when the event store and business database are separate.
• Prefer commands over direct event publishing for domain operations — commands express intent, events record facts.
• Use middleware for cross-cutting concerns like logging, metrics, and validation rather than mixing them into handlers.