API Reference

Protocols

`CommandBusProtocol`

Protocol for command bus implementations.

Example

class CommandBusProtocol:
    async def dispatch(self, command: Command) -> Any:
        handler = self._handlers[type(command)]
        return await handler.handle(command)

dispatch

async def dispatch(command: Any) -> Any

source

Dispatch a command to its handler.

Parameters

Parameter	Type	Description
`command`	Any	Command to dispatch.

Returns

Type	Description
Any	Result from the command handler.

`DomainEventPublisherProtocol`

source

Protocol for publishing domain events.

publish

async def publish(event: Any) -> None

source

Publish a domain event.

Parameters

Parameter	Type	Description
`event`	Any	DomainEvent instance.

`EventBusProtocol`

source

Protocol for event bus implementations.

The event bus manages event publication and subscription.

Example

```python
class InMemoryEventBus:
    async def publish(self, event: DomainEvent) -> "Result[None, EventError]":
        for handler in self._handlers.get(type(event), []):
            await handler.handle(event)
        return Ok(None)

    def subscribe(self, event_type, handler):
        self._handlers.setdefault(event_type, []).append(handler)

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>publish</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>publish</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'>Any</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>Result<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventerror'>EventError</a><span class='qk-c'>]</span></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L87' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Publish an event to all subscribers.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event`</td><td class='qk-tt'>Any</td><td class='qk-td'>DomainEvent to publish.</td></tr></tbody></table></div>




<div class='qk-table-wrap'><span class='qk-label'>Returns</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-45'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tt'>Result<span class='qk-c'>[</span><span class='qk-d'>None</span><span class='qk-c'>,</span> <a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventerror'>EventError</a><span class='qk-c'>]</span></td><td class='qk-td'>Ok(None) when the event is successfully enqueued for dispatch. Err(EventError) when the event cannot be accepted (e.g.\ no handlers registered and the bus requires at least one).</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>subscribe</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>subscribe</span><span class='qk-c'>(</span>
    <span class='qk-n'>event_type</span><span class='qk-c'>: </span><span class='qk-t'>type</span><span class='qk-c'>,</span>
    <span class='qk-n'>handler</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L100' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Subscribe a handler to an event type.

Implementations accept either an ``EventHandlerProtocol`` object or a
plain async callable — the latter is what ``@event_handler`` produces
and what the flagship
EventBus dispatches directly.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event_type`</td><td class='qk-tt'>type</td><td class='qk-td'>Type of event to subscribe to.</td></tr><tr><td class='qk-tn'>`handler`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></td><td class='qk-td'>Handler to call when event is published.</td></tr></tbody></table></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>flush</span></div><pre class='qk-sig-pre'><span class='qk-k'>async </span><span class='qk-k'>def </span><span class='qk-f'>flush</span><span class='qk-c'>(</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L118' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Wait until all pending events have been dispatched.

Useful in tests and during shutdown to drain in-flight handler
dispatches before tearing the bus down.  Implemented by the flagship
EventBus; engines that dispatch
synchronously may implement it as a no-op.

<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>unsubscribe</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>unsubscribe</span><span class='qk-c'>(</span>
    <span class='qk-n'>event_type</span><span class='qk-c'>: </span><span class='qk-t'>type</span><span class='qk-c'>,</span>
    <span class='qk-n'>handler</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></span>
<span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-d'>None</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/domain.py#L128' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>

Remove a handler subscription for an event type.



<div class='qk-table-wrap'><span class='qk-label'>Parameters</span><table class='qk-table'><thead><tr><th class='qk-th qk-w-20'>Parameter</th><th class='qk-th qk-w-25'>Type</th><th class='qk-th qk-th--desc qk-w-55'>Description</th></tr></thead><tbody><tr><td class='qk-tn'>`event_type`</td><td class='qk-tt'>type</td><td class='qk-td'>Type of event to unsubscribe from.</td></tr><tr><td class='qk-tn'>`handler`</td><td class='qk-tt'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#eventhandlerprotocol'>EventHandlerProtocol</a> <span class='qk-c'><span class='qk-c'>|</span></span> Callable<span class='qk-c'>[</span>Ellipsis<span class='qk-c'>,</span> Awaitable<span class='qk-c'>[</span>Any<span class='qk-c'>]</span><span class='qk-c'>]</span></td><td class='qk-td'>Handler to remove.</td></tr></tbody></table></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `EventFilterProtocol`

</div>

<span class='qk-meta' data-api-type='Protocols'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-events/src/quadkit/events/protocols.py#L22' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Predicate that determines whether an event should be dispatched.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>matches</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>matches</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#domainevent'>DomainEvent</a></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bool</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-events/src/quadkit/events/protocols.py#L25' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `EventSerializerProtocol`

</div>

<span class='qk-meta' data-api-type='Protocols'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-events/src/quadkit/events/protocols.py#L14' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Serializes and deserializes domain events for transport.

<div class='qk-methods'>
<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>serialize</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>serialize</span><span class='qk-c'>(</span><span class='qk-n'>event</span><span class='qk-c'>: </span><span class='qk-t'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#domainevent'>DomainEvent</a></span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'>bytes</span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-events/src/quadkit/events/protocols.py#L17' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>



<div class='qk-sig qk-sig--compact'><div class='qk-sig-head'><span class='qk-sig-label'>deserialize</span></div><pre class='qk-sig-pre'><span class='qk-k'>def </span><span class='qk-f'>deserialize</span><span class='qk-c'>(</span><span class='qk-n'>data</span><span class='qk-c'>: </span><span class='qk-t'>bytes</span><span class='qk-c'>)</span><span class='qk-k'> -> </span><span class='qk-r'><a class='qk-typelink' href='/quadkit/packages/events/quadkit-events/api/#domainevent'>DomainEvent</a></span></pre></div>

<div class='qk-source-row'><a class='qk-source qk-source--sm' href='https://github.com/dbtinoy-/quadkit/blob/main/packages/quadkit-events/src/quadkit/events/protocols.py#L18' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>


</div>

<hr class='qk-rule' />

<div data-pagefind-weight='10'>

### `EventStoreProtocol`

</div>

<span class='qk-meta' data-api-type='Protocols'></span>

<div class='qk-source-row'><a class='qk-source' href='https://github.com/dbtinoy-/quadkit/blob/main/core/quadkit-contracts/src/quadkit/contracts/events/protocols/store.py#L14' target='_blank' rel='noopener noreferrer'><span class='qk-source-icon' aria-hidden='true'></span>source</a></div>
Protocol for event store implementations.

The event store persists and retrieves domain events.


**Example**

```python
class PostgresEventStore:
    async def append(self, stream_id, events, expected_version=None):
        async with self.db.transaction():
            for event in events:
                await self.db.insert("events", event.to_dict())

append

async def append(
    stream_id: str,
    events: list[Any],
    expected_version: int | None = None
) -> int

source

Append events to a stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique stream identifier.
`events`	list[Any]	List of events to append.
`expected_version`	int \| None	Expected stream version for optimistic concurrency.

Returns

Type	Description
int	New stream version.

Raises

Exception	Description
ConcurrencyError	If expected version doesn't match.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Any]

source

Read events from a stream.

Every shipped store — abstract, in-memory, SQLite, Postgres, MongoDB, Redis, outbox and streaming — declares this exact signature. The protocol previously said (stream_id, start, count); because the container only checks member presence, that drift bound cleanly and surfaced only as type: ignore[call-arg] at call sites (S23). Widening here is backward compatible: the added parameter is optional, so positional and keyword callers both keep working.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique stream identifier.
`from_version`	int	First version to return; 0 reads from the start.
`to_version`	int \| None	Last version to return, inclusive; ``None`` is unbounded.
`limit`	int \| None	Maximum number of events to return.

Returns

Type	Description
list[Any]	List of events.

read_all

async def read_all(
    position: int = 0,
    count: int | None = None
) -> list[Any]

source

Read events from all streams.

Parameters

Parameter	Type	Description
`position`	int	Starting global position.
`count`	int \| None	Maximum events to read.

Returns

Type	Description
list[Any]	List of events.

`MultiEventHandlerProtocol`

source

Protocol for handlers that handle multiple event types.

handles

def handles() -> list[type]

source

Get list of event types handled.

Returns

Type	Description
list[type]	List of event classes.

handle

async def handle(event: Any) -> Result[None, EventError]

source

Handle an event.

Parameters

Parameter	Type	Description
`event`	Any	DomainEvent to handle.

Returns

Type	Description
Result[None, EventError]	``Ok(None)` `on success,` `Err(EventError)`` if handling fails in an expected, recoverable way.

`ProjectionProtocol`

source

ProjectionProtocol protocol for building read models from events.

apply

def apply(event: Any) -> None

source

Apply an event to the projection state.

Parameters

Parameter	Type	Description
`event`	Any	Domain event to apply.

`QueryBusProtocol`

source

Protocol for query bus implementations.

Example

class QueryBusProtocol:
    async def execute(self, query: Query) -> Any:
        handler = self._handlers[type(query)]
        return await handler.handle(query)

execute

async def execute(query: Any) -> Any

source

Execute a query through its handler.

Parameters

Parameter	Type	Description
`query`	Any	Query to execute.

Returns

Type	Description
Any	Query result.

`ReadOnlyRepositoryProtocol`

source

Protocol for read-only repository operations.

Use this for query-only access patterns (CQRS query side).

Example

class UserQueryRepository:
    async def get(self, id: str) -> User | None:
        return await self.db.query("users").where(id=id).first()

    async def list(self, skip: int = 0, limit: int = 100) -> list[User]:
        return await self.db.query("users").offset(skip).limit(limit).all()

get

async def get(item_id: str) -> T | None

source

Get entity by ID.

Parameters

Parameter	Type	Description
`item_id`	str	Entity identifier.

Returns

Type	Description
T \| None	Entity if found, None otherwise.

list

async def list(
    skip: int = 0,
    limit: int = 100,
    **filters: Any
) -> list[T]

source

List entities with pagination.

Parameters

Parameter	Type	Description
`skip`	int	Number of records to skip.
`limit`	int	Maximum records to return. **filters: Optional filter criteria.

Returns

Type	Description
list[T]	List of entities.

find_by_spec

async def find_by_spec(spec: SpecificationProtocol[T]) -> list[T]

source

Find entities matching a complex specification.

Parameters

Parameter	Type	Description
`spec`	SpecificationProtocol[T]	The DDD specification to evaluate.

Returns

Type	Description
list[T]	List of matching entities.

count

async def count(**filters: Any) -> int

source

Count entities matching filters.

Parameters

Parameter	Type	Description

Returns

Type	Description
int	Total count of matching entities.

`RepositoryProtocol`

source

Protocol for full repository operations.

Extends ReadOnlyRepositoryProtocol with write operations.

Example

class UserRepository:
    async def save(self, entity: User) -> User:
        if entity.id:
            await self.db.update("users", entity.dict()).where(id=entity.id)
        else:
            entity.id = await self.db.insert("users", entity.dict())
        return entity

    async def delete(self, id: str) -> bool:
        result = await self.db.delete("users").where(id=id)
        return result.affected_rows > 0

save

async def save(entity: T) -> T

source

Save (create or update) an entity.

Parameters

Parameter	Type	Description
`entity`	T	Entity to save.

Returns

Type	Description
T	Saved entity with any generated fields populated.

delete

async def delete(item_id: str) -> bool

source

Delete entity by ID.

Parameters

Parameter	Type	Description
`item_id`	str	Entity identifier.

Returns

Type	Description
bool	True if deleted, False if not found.

save_many

async def save_many(entities: list[T]) -> list[T]

source

Save (create or update) multiple entities in a single operation.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
`entities`	list[T]	Entities to save.

Returns

Type	Description
list[T]	Saved entities with any generated fields populated.

delete_many

async def delete_many(item_ids: list[str]) -> int

source

Delete multiple entities by ID.

Implementations SHOULD execute this as a batch for efficiency.

Parameters

Parameter	Type	Description
`item_ids`	list[str]	Entity identifiers to delete.

Returns

Type	Description
int	Number of entities actually deleted.

`SagaManagerProtocol`

source

Protocol for saga lifecycle management.

The manager routes incoming events to all registered sagas and coordinates their execution.

process

async def process(event: Any) -> None

source

Process an event through all relevant sagas.

Parameters

Parameter	Type	Description
`event`	Any	Domain event to route and process.

`SagaProtocol`

source

Protocol for saga / process-manager implementations.

Sagas coordinate long-running business processes that span multiple aggregates by reacting to domain events and dispatching commands.

handle

async def handle(event: Any) -> list[Any]

source

Handle a domain event and produce commands.

Parameters

Parameter	Type	Description
`event`	Any	Domain event to handle.

Returns

Type	Description
list[Any]	List of commands to dispatch to the command bus.

`SnapshotStoreProtocol`

source

Protocol for aggregate snapshot storage.

Snapshots are stored as objects, not as loose tuples: every implementation in the framework — InMemorySnapshotStore and the Postgres, SQLite and MongoDB stores — exposes save_snapshot/get_latest/get_by_version/ delete_old_snapshots, and SnapshotManager, the only consumer, calls exactly those.

This declaration used to be a pre-migration tuple API (save(aggregate_id, snapshot, version) / load(aggregate_id)) that no implementation ever had. It was not harmless: the container validates protocol conformance at registration, so binding a store to this protocol failed with QK_ERR_DI_007 — which is why the framework’s own EventsModule could export a protocol that could never be resolved, and why SnapshotManager had to silence three attribute errors with type: ignore.

save_snapshot

async def save_snapshot(snapshot: Any) -> None

source

Save an aggregate snapshot.

Parameters

Parameter	Type	Description
`snapshot`	Any	Snapshot object carrying the aggregate id, version and state. Typed ``Any` `because the snapshot type lives in` `quadkit-events`` and contracts must not depend on packages.

get_latest

async def get_latest(aggregate_id: str) -> Any | None

source

Load the latest snapshot for an aggregate.

Parameters

Parameter	Type	Description
`aggregate_id`	str	Aggregate identifier.

Returns

Type	Description
Any \| None	The newest snapshot, or ``None`` when the aggregate has none.

get_by_version

async def get_by_version(
    aggregate_id: str | UUID,
    version: int
) -> Any | None

source

Load one specific snapshot version.

Parameters

Parameter	Type	Description
`aggregate_id`	str \| UUID	Aggregate identifier.
`version`	int	Exact snapshot version.

Returns

Type	Description
Any \| None	The snapshot at that version, or ``None``.

delete_old_snapshots

async def delete_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int = 3
) -> int

source

Delete superseded snapshots, keeping the newest keep_count.

Parameters

Parameter	Type	Description
`aggregate_id`	str \| UUID	Aggregate identifier.
`keep_count`	int	Number of newest snapshots to retain.

Returns

Type	Description
int	Number of snapshots deleted.

async def close() -> None

source

Release the store’s resources.

Classes

`AbstractEventStore`

source

Abstract base class for event storage.

Inherits stream/cursor/replay helpers from EventStreamMixin. Snapshot storage is provided by AbstractSnapshotStore.

Example

# Append events
await store.append(
    stream_id="order-123",
    events=[OrderCreated(...), ItemAdded(...)],
    expected_version=0
)

# Read events
events = await store.read("order-123")

# Append events
await store.append(
    stream_id="order-123",
    events=[OrderCreated(...), ItemAdded(...)],
    expected_version=0
)

# Read events
events = await store.read("order-123")

__init__

def __init__(schema_evolution: SchemaEvolution | None = None) -> None

source

Initialise event store, optionally with schema evolution (M-03).

append

async def append(
    stream_id: str,
    events: list[Event],
    expected_version: int | None = None
) -> int

source

Append events to a stream with optimistic concurrency.

Returns

Type	Description
int	New stream version.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique identifier for the event stream
`events`	list[Event]	List of events to save
`expected_version`	int \| None	Expected current version for concurrency control

Raises

Exception	Description
ConcurrencyError	If version conflict occurs
EventPersistenceError	If save operation fails

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Event]

source

Read events from a stream (M-15).

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to load events from
`from_version`	int	Starting version (inclusive)
`to_version`	int \| None	Ending version (inclusive)
`limit`	int \| None	Max number of events to read

Returns

Type	Description
list[Event]	List of events in chronological order

Raises

Exception	Description
EventLoadError	If load operation fails

get_stream_version

async def get_stream_version(stream_id: str) -> int

source

Get the current version of a stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to check

Returns

Type	Description
int	Current version (0 if stream doesn't exist)

stream_exists

async def stream_exists(stream_id: str) -> bool

source

Check if a stream exists.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to check

Returns

Type	Description
bool	True if stream has events

get_stream_info

async def get_stream_info(stream_id: str) -> StreamInfo | None

source

Get information about a stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to get info for

Returns

Type	Description
StreamInfo \| None	StreamInfo or None if stream doesn't exist

read_all

async def read_all(
    position: int = 0,
    count: int | None = None
) -> list[Event]

source

Read all events from the store in global order.

Default implementation streams from the store’s stream_all primitive so every backend inherits a working read_all instead of the no-op stub leaked by EventStoreProtocol.

Parameters

Parameter	Type	Description
`position`	int	Starting global sequence number (inclusive).
`count`	int \| None	Maximum number of events to return; ``None`` for all.

Returns

Type	Description
list[Event]	List of events in global order, newest last.

stream_all

def stream_all(
    from_position: int = 0,
    batch_size: int = 100,
    partition: int | None = None,
    total_partitions: int | None = None
) -> AsyncIterator[Event]

source

Stream all events from the store in global order.

Parameters

Parameter	Type	Description
`from_position`	int	Starting global sequence number.
`batch_size`	int	Number of events to fetch per batch.
`partition`	int \| None	Optional partition index (0 to total_partitions - 1).
`total_partitions`	int \| None	Total number of partitions for sharding.

compact

async def compact(
    stream_id: str,
    up_to_version: int
) -> int

source

Purge events for a stream up to (and including) the given version (MF-04).

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.
`up_to_version`	int	Maximum version to delete.

Returns

Type	Description
int	Number of events purged.

find_by_aggregate

async def find_by_aggregate(aggregate_id: str) -> list[Event]

source

Get all events for an aggregate (for reconstitution).

Parameters

Parameter	Type	Description
`aggregate_id`	str	Aggregate identifier (same as stream_id).

Returns

Type	Description
list[Event]	All events for the aggregate in chronological order.

async def close() -> None

source

Close the event store and release resources.

`AbstractMiddleware`

source

Abstract base for middleware components.

Middleware wraps around handlers to add cross-cutting concerns like logging, validation, transactions, etc.

Example

class TimingMiddleware(AbstractMiddleware[Message, Any]):
    async def __call__(
        self,
        message: Message,
        next_handler: NextHandler
    ) -> Any:
        start = time.time()
        result = await next_handler(message)
        duration = time.time() - start
        logger.info("Handler took %.3fs", duration)
        return result

`AbstractReadOnlyRepository`

source

Read-only repository interface.

Useful for query-side repositories in CQRS.

get

async def get(aggregate_id: UUID | str) -> TAggregate | None

source

Get an aggregate by ID.

exists

async def exists(aggregate_id: UUID | str) -> bool

source

Check if an aggregate exists.

get_all

async def get_all(
    limit: int | None = None,
    offset: int | None = None
) -> list[TAggregate]

source

Get all aggregates with pagination.

`AbstractRepository`

source

Abstract base repository for aggregate persistence.

Repositories provide collection-like access to aggregates, abstracting away the persistence mechanism.

Example

    async def save(self, aggregate: Order) -> None:
        ...
```python

get

async def get(aggregate_id: UUID | str) -> TAggregate | None

source

Get an aggregate by ID.

Parameters

Parameter	Type	Description
`aggregate_id`	UUID \| str	The aggregate identifier

Returns

Type	Description
TAggregate \| None	The aggregate if found, None otherwise

save

async def save(aggregate: TAggregate) -> None

source

Save an aggregate.

This persists any pending changes and clears the pending events.

Parameters

Parameter	Type	Description
`aggregate`	TAggregate	The aggregate to save

exists

async def exists(aggregate_id: UUID | str) -> bool

source

Check if an aggregate exists.

Parameters

Parameter	Type	Description
`aggregate_id`	UUID \| str	The aggregate identifier

Returns

Type	Description
bool	True if the aggregate exists

`AbstractSnapshotStore`

source

Abstract base class for snapshot storage.

Provides optimised aggregate loading by storing periodic snapshots. Avoids replaying the entire event history for aggregates with many events.

Strategy: 1. Load the latest snapshot (if any). 2. Load events after the snapshot version. 3. Apply those events to the snapshot state.

This reduces loading from O(N) to O(1) + O(events since snapshot).

Example (M-14 — uses Snapshot object throughout)

from quadkit.events.types import Snapshot

# Save a snapshot
await store.save(Snapshot(
    aggregate_id=order.id,
    aggregate_type="Order",
    version=order.version,
    state=order.to_dict()
))

# Load the latest snapshot
snapshot = await store.get_latest("order-123")
if snapshot:
    order = Order.from_snapshot(snapshot.state)
    events = await event_store.read("order-123", from_version=snapshot.version)

from quadkit.events.types import Snapshot

# Save a snapshot
await store.save(Snapshot(
    aggregate_id=order.id,
    aggregate_type="Order",
    version=order.version,
    state=order.to_dict()
))

# Load the latest snapshot
snapshot = await store.get_latest("order-123")
if snapshot:
    order = Order.from_snapshot(snapshot.state)
    events = await event_store.read("order-123", from_version=snapshot.version)

save_snapshot

async def save_snapshot(snapshot: Snapshot) -> None

source

Save an aggregate snapshot.

Parameters

Parameter	Type	Description
`snapshot`	Snapshot	Snapshot object containing aggregate_id, version, and state.

get_latest

async def get_latest(aggregate_id: str) -> Snapshot | None

source

Load the latest snapshot for an aggregate.

Parameters

Parameter	Type	Description
`aggregate_id`	str	Aggregate to load snapshot for.

Returns

Type	Description
Snapshot \| None	Snapshot object or None if no snapshot exists.

get_by_version

async def get_by_version(
    aggregate_id: str | UUID,
    version: int
) -> Snapshot | None

source

Get a specific snapshot by version.

Parameters

Parameter	Type	Description
`aggregate_id`	str \| UUID	Aggregate identifier.
`version`	int	Exact version to retrieve.

Returns

Type	Description
Snapshot \| None	Snapshot at that version, or None.

delete_old_snapshots

async def delete_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int = 3
) -> int

source

Delete old snapshots, keeping only the most recent ones.

Parameters

Parameter	Type	Description
`aggregate_id`	str \| UUID	ID of the aggregate.
`keep_count`	int	Number of snapshots to retain (default: 3).

Returns

Type	Description
int	Number of snapshots deleted.

async def close() -> None

source

Close the snapshot store and release resources.

`AggregateFactoryProtocol`

source

Factory for creating aggregates.

create

def create(aggregate_id: UUID | str) -> TAggregate_co

source

Create a new aggregate instance.

Parameters

Parameter	Type	Description
`aggregate_id`	UUID \| str	Aggregate ID.

Returns

Type	Description
TAggregate_co	New aggregate instance.

`AggregateRoot`

source

Base class for event-sourced aggregates.

Aggregates:

Maintain consistency boundaries
Generate domain events on state changes
Can be reconstructed from events (event sourcing)
Support snapshotting for performance

Example

class Order(AggregateRoot):
    status: str = "draft"
    items: list[OrderItem] = Field(default_factory=list)
    total: Decimal = Decimal("0.00")

    def add_item(self, product_id: str, quantity: int, price: Decimal):
        if self.status != "draft":
            raise InvalidOperationError("Cannot modify non-draft order")

        self._apply(ItemAddedEvent(
            order_id=self.id,
            product_id=product_id,
            quantity=quantity,
            price=price
        ))

    def _handle_item_added(self, event: ItemAddedEvent):
        self.items.append(OrderItem(
            product_id=event.product_id,
            quantity=event.quantity,
            price=event.price
        ))
        self.total += event.price * event.quantity

__init__

def __init__(
    id: UUID | None = None,
    **kwargs: Any
) -> None

source

Initialise the aggregate, auto-generating a UUID id if none is given.

add_event

def add_event(event: Any) -> None

source

Buffer a domain event for persistence, calling invariant checks first.

pull_events

def pull_events() -> list[Any]

source

Return a copy of pending events without clearing them.

clear_events

def clear_events() -> None

source

Discard all buffered events.

has_uncommitted_events

property has_uncommitted_events() -> bool

source

True when there are buffered events awaiting persistence.

get_pending_events

def get_pending_events() -> list[Event]

source

Get events pending persistence.

Returns

Type	Description
list[Event]	List of uncommitted events

clear_pending_events

def clear_pending_events() -> None

source

Clear pending events after persistence.

load_from_history

def load_from_history(events: list[Event]) -> None

source

Reconstruct aggregate state from event history.

This is used during aggregate loading to replay all events and rebuild the current state.

Parameters

Parameter	Type	Description
`events`	list[Event]	Historical events to replay

load_from_snapshot

def load_from_snapshot(snapshot: Snapshot) -> None

source

Restore aggregate state from a snapshot.

Parameters

Parameter	Type	Description
`snapshot`	Snapshot	The snapshot to restore from

create_snapshot

def create_snapshot() -> Snapshot

source

Create a snapshot of current aggregate state.

Returns

Type	Description
Snapshot	Snapshot instance

`AggregateStatus`

source

Status of an aggregate root.

`Checkpoint`

source

Checkpoint for tracking projection progress.

`Command`

source

Command extended with aggregate-targeting helpers.

target_aggregate_id

property target_aggregate_id() -> UUID | None

source

expected_version

property expected_version() -> int | None

source

for_aggregate

def for_aggregate(
    aggregate_id: UUID,
    version: int | None = None
) -> Command

source

Target a specific aggregate.

`CommandBusConfig`

source

Configuration for command bus.

`CommandBusImpl`

source

Command bus for dispatching commands.

Handlers are registered explicitly via register() or through HandlerRegistry.register_with_buses() during provider registration.

Example

# Commands are automatically routed to handlers
result = await bus.dispatch(CreateOrderCommand(
    customer_id="cust-123",
    items=[...]
))

# Commands are automatically routed to handlers
result = await bus.dispatch(CreateOrderCommand(
    customer_id="cust-123",
    items=[...]
))

__init__

def __init__(
    middlewares: list[Any] | None = None,
    config: Any | None = None
) -> None

source

Initialize the command bus.

dispatch

async def dispatch(command: Command) -> Any

source

Dispatch a command for execution.

Parameters

Parameter	Type	Description
`command`	Command	The command to execute.

Returns

Type	Description
Any	Command execution result.

Raises

Exception	Description
HandlerNotFoundError	If no handler is registered.
CommandExecutionError	If command execution fails.

`CommandHandlerProtocol`

source

Base class for command handlers.

Command handlers process commands and optionally return results. Each command type should have exactly one handler.

Example

class CreateOrderHandler(CommandHandlerProtocol[CreateOrderCommand, OrderId]):
    def __init__(self, repository: OrderRepository):
        self.repository = repository

    async def handle(self, command: CreateOrderCommand) -> OrderId:
        order = Order.create(
            customer_id=command.customer_id,
            items=command.items
        )
        await self.repository.save(order)
        return order.id

handle

async def handle(command: TCommand) -> TResult

source

Handle a command.

Parameters

Parameter	Type	Description
`command`	TCommand	The command to handle

Returns

Type	Description
TResult	Command execution result

`CommandResult`

source

Result of command execution.

`DispatchResult`

source

Result of an event dispatch operation.

Attributes: success: True if all handlers completed successfully. handler_results: List of return values from each handler. errors: List of (exception_type, exception_instance) tuples.

raise_if_errors

def raise_if_errors() -> None

source

Raise if any handler failed. Call explicitly when needed.

has_errors

property has_errors() -> bool

source

Return True if any handler failed.

`DomainEvent`

source

Base class for domain events.

__init__

def __init__(**kwargs: Any) -> None

source

Initialize the event setting base fields and any extra subclass fields.

Declared dataclass fields receive their default values when not supplied. Extra keyword arguments (e.g. fields declared on plain subclasses without @dataclass) are set directly on the instance via object.__setattr__, which bypasses the frozen=True guard safely during construction.

to_dict

def to_dict() -> dict[str, Any]

source

Return a JSON-serializable dictionary representation.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> DomainEvent

source

Rebuild an event from to_dict output.

The inverse of to_dict for the base fields it stringified: a UUID that travelled as text comes back as a UUID, a timestamp as a datetime. Anything else is handed to the constructor unchanged, so a round trip through an outbox table or a JSON log yields an event that compares field-for-field with the one that was stored.

for_aggregate

def for_aggregate(
    aggregate_id: UUID,
    aggregate_type: str
) -> DomainEvent

source

`Entity`

source

Base class for entities within aggregates.

Entities:

Have identity (id field)
Are mutable
Belong to an aggregate root
Don’t have their own event streams

touch

def touch() -> None

source

Update the modified timestamp.

`Event`

source

Represents a fact that happened in the past.

We override __init__ to accept arbitrary extra keyword arguments, which are attached as attributes on the instance. This preserves the historical ability to create Event(data=...) without declaring a new subclass. event_type handling is performed by the contract base class (which now includes actor_id).

The event version is an aggregate-specific concept and therefore is declared on this subclass rather than the canonical DomainEvent base class. with_version and related helpers operate on this field.

__init__

def __init__(
    *args: Any,
    **kwargs: Any
) -> None

source

with_version

def with_version(version: int) -> Event

source

Create a copy with a specific version (aggregate version).

for_aggregate

def for_aggregate(
    aggregate_id: UUID,
    aggregate_type: str
) -> Event

source

Create a copy for a specific aggregate.

`EventBusConfig`

source

Configuration for event bus.

`EventBusImpl`

source

Event bus for publishing domain events.

Orchestrates subscription management, per-event-type BoundedChannel backpressure (MAJ-11), and background drain tasks. All handler dispatch logic (lookup, parallel/sequential dispatch, retry) is provided by HandlerDispatchMixin.

Attributes: DEFAULT_MAX_CONCURRENT_HANDLERS: Maximum concurrent handler executions. DEFAULT_HANDLER_TIMEOUT_SECONDS: Timeout for each handler execution. DEFAULT_RETRY_FAILED_HANDLERS: Whether to retry failed handlers. DEFAULT_MAX_HANDLER_RETRIES: Maximum retry attempts per handler. DEFAULT_ENABLE_DEAD_LETTER: Whether to enable dead letter queue. DEFAULT_ALLOW_NO_HANDLERS: Whether to allow events with no handlers. DEFAULT_PARALLEL_DISPATCH: Whether to dispatch handlers in parallel. DEFAULT_CONTINUE_ON_ERROR: Whether to continue on handler errors.

__init__

def __init__(
    middlewares: list[MiddlewareFunc] | None = None,
    config: EventBusConfig | None = None,
    parallel: ParallelProtocol | None = None,
    tracer: TracerProtocol | None = None,
    hooks: HookRegistryProtocol | None = None
) -> None

source

Initialize the event bus.

Parameters

Parameter	Type	Description
`middlewares`	list[MiddlewareFunc] \| None	List of middleware functions.
`config`	EventBusConfig \| None	Event bus configuration.
`parallel`	ParallelProtocol \| None	Parallel execution protocol for concurrent handler execution.
`tracer`	TracerProtocol \| None	Optional tracer for distributed tracing.

dispatch_errors

property dispatch_errors() -> tuple[Exception, Ellipsis]

source

Return handler failures observed during asynchronous dispatch.

publish() reports enqueue acceptance, while handlers run in the background drain. This read-only snapshot lets health checks and operator surfaces inspect failures even when continue_on_error is enabled.

clear_dispatch_errors

def clear_dispatch_errors() -> None

source

Clear the observed dispatch failures after an operator has read them.

def subscribe(
    event_type: type[Event],
    handler: Any
) -> None

source

Subscribe a handler to an event type.

Parameters

Parameter	Type	Description
`event_type`	type[Event]	The event class to subscribe to.
`handler`	Any	Handler to call when event is published.

set_tracer

def set_tracer(tracer: TracerProtocol | None) -> None

source

Attach an optional tracer after provider boot wiring.

Parameters

Parameter	Type	Description
`tracer`	TracerProtocol \| None	Tracer to attach, or None to clear.

set_hook_registry

def set_hook_registry(hooks: HookRegistryProtocol | None) -> None

source

Attach an optional hook registry after provider boot wiring.

handler

def handler(event_type: type[Event]) -> Callable[[Callable], Callable]

source

Decorator to subscribe a handler to an event type.

subscribe_all

def subscribe_all(handler: Any) -> None

source

Subscribe a handler to all events.

Parameters

Parameter	Type	Description
`handler`	Any	Handler to call for every published event.

unsubscribe

def unsubscribe(
    event_type: type[Event],
    handler: Any
) -> bool

source

Unsubscribe a handler from an event type.

Parameters

Parameter	Type	Description
`event_type`	type[Event]	The event class.
`handler`	Any	Handler to remove.

Returns

Type	Description
bool	True if handler was found and removed.

publish

async def publish(event: Event) -> Result[None, EventError]

source

Publish an event to all subscribers via a bounded channel.

Enqueues the event into a per-event-type BoundedChannel; a background drain task dequeues and dispatches it to all handlers. Blocks when the channel is at capacity providing natural backpressure (MAJ-11).

Parameters

Parameter	Type	Description
`event`	Event	The event to publish.

Returns

Type	Description
Result[None, EventError]	Ok(None) on successful enqueue. Err(EventHandlerError) when no handlers are registered and allow_no_handlers is False. Err(EventError) once the bus has been closed: a publish after shutdown must not quietly start a new drain task.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Report event-bus readiness and accumulated dispatch failures.

Publication is intentionally asynchronous, so a successful publish() does not prove that every handler completed. The diagnostics count makes that distinction visible to health consumers.

async def close(timeout: float = 30.0) -> None

source

Drain what was accepted, then stop the background drain tasks.

publish() returns as soon as the event is queued, so at shutdown the queue may still hold events the caller believes are delivered. Closing each channel lets its drain task finish the queued events and exit; tasks that are still running after timeout seconds are cancelled. Idempotent: a second call is a no-op.

Only drain tasks owned by the running loop are waited on; a task started on another loop is left to settle there, because its completion can never be observed from here.

Parameters

Parameter	Type	Description
`timeout`	float	Seconds to wait for queued events to be handled before the remaining drain tasks are cancelled.

flush

async def flush() -> None

source

Wait until all pending events in every channel have been dispatched.

Yields control repeatedly until every BoundedChannel is empty and all in-progress handler dispatches have completed. Useful in tests.

`EventEnvelope`

source

Wrapper for events with metadata for storage and transmission.

`EventHandledHook`

source

Payload fired after an event handler successfully processes an event.

Attributes: event_type: Fully-qualified type name of the handled event. handler: Qualified name of the handler class or function.

`EventHandlerProtocol`

source

Base class for event handlers.

Event handlers process events and perform side effects. Unlike command handlers, multiple handlers can subscribe to the same event type.

Example

class SendOrderConfirmationHandler(EventHandlerProtocol[OrderCreatedEvent]):
    def __init__(self, email_service: EmailService):
        self.email_service = email_service

    async def handle(self, event: OrderCreatedEvent) -> None:
        await self.email_service.send_confirmation(
            order_id=event.order_id,
            customer_email=event.customer_email
        )

handle

async def handle(event: TEvent) -> Result[None, EventError]

source

Handle an event.

Parameters

Parameter	Type	Description
`event`	TEvent	The event to handle.

Returns

Type	Description
Result[None, EventError]	``Ok(None)` `on success,` `Err(EventError)`` if handling fails in an expected, recoverable way.

`EventPublishedEvent`

source

Emitted when the event bus successfully publishes a domain event.

Attributes: source_event_type: Fully-qualified type name of the published event. handler_count: Number of registered handlers that were notified.

`EventPublishedHook`

source

Payload fired when an event is published onto the event bus.

Attributes: event_type: Fully-qualified type name of the published event. aggregate_id: Identifier of the aggregate that emitted the event, if known.

`EventSourcingRepository`

source

AbstractRepository using event sourcing with snapshot acceleration.

This repository:

Loads aggregates using snapshots when available
Persists changes as events
Automatically creates snapshots based on policy
Publishes events to the event bus

Example

# Create repository with snapshot support
repository = EventSourcingRepository(
    aggregate_type=Order,
    event_store=event_store,
    snapshot_store=snapshot_store,
    event_bus=event_bus,
    snapshot_policy=EventCountPolicy(100)
)

# Load aggregate - uses snapshot if available
order = await repository.get(order_id)

# Modify aggregate
order.add_item(product_id, quantity, price)

# Save - stores events, maybe creates snapshot, publishes events
await repository.save(order)

Performance depends on the backend, snapshot size and remaining events.

__init__

def __init__(
    aggregate_type: type[TAggregate],
    event_store: AbstractEventStore,
    snapshot_store: AbstractSnapshotStore | None = None,
    event_bus: EventBusImpl | None = None,
    snapshot_policy: SnapshotPolicy | None = None,
    snapshot_every: int = 100
)

source

Initialize the event sourcing repository.

Parameters

Parameter	Type	Description
`aggregate_type`	type[TAggregate]	The aggregate class to manage
`event_store`	AbstractEventStore	Event store for event persistence
`snapshot_store`	AbstractSnapshotStore \| None	Snapshot store (optional, avoids replaying earlier events)
`event_bus`	EventBusImpl \| None	Event bus for publishing events (optional)
`snapshot_policy`	SnapshotPolicy \| None	Custom snapshot policy (default: every N events)
`snapshot_every`	int	Events between snapshots (default: 100)

get

async def get(aggregate_id: UUID | str) -> TAggregate | None

source

Load an aggregate; reject snapshots or replay that cannot be applied.

save

async def save(aggregate: TAggregate) -> None

source

Save an aggregate’s pending events.

This method:

Persists pending events to the event store
Creates a snapshot if policy triggers
Publishes events to the event bus
Clears pending events from the aggregate

Parameters

Parameter	Type	Description
`aggregate`	TAggregate	The aggregate to save

Raises

Exception	Description
ConcurrencyError	If version conflict detected

exists

async def exists(aggregate_id: UUID | str) -> bool

source

Check if an aggregate exists.

Parameters

Parameter	Type	Description
`aggregate_id`	UUID \| str	The aggregate identifier

Returns

Type	Description
bool	True if the event stream has a positive logical version. This does not establish that the aggregate can be replayed or that a snapshot without events exists.

get_or_raise

async def get_or_raise(aggregate_id: UUID | str) -> TAggregate

source

Get an aggregate or raise an error if not found.

Parameters

Parameter	Type	Description
`aggregate_id`	UUID \| str	The aggregate identifier

Returns

Type	Description
TAggregate	The aggregate

Raises

Exception	Description
AggregateNotFoundError	If aggregate not found

force_snapshot

async def force_snapshot(aggregate: TAggregate) -> None

source

Force creation of a snapshot for an aggregate.

Useful for bulk operations or maintenance.

Parameters

Parameter	Type	Description
`aggregate`	TAggregate	The aggregate to snapshot

`EventStoreBackend`

source

Supported event store backends.

`EventStoredHook`

source

Payload fired when an event is persisted to the event store.

Attributes: event_type: Fully-qualified type name of the stored event. stream_id: Identifier of the event stream the event was written to.

`EventsConfig`

source

Top-level Events configuration.

This consolidated class combines all event system configuration. No TYPE_CHECKING guards or cast() wrappers — all classes defined in this module, so runtime type resolution and dict→model coercion work.

validate_backend_config

def validate_backend_config() -> list[str]

source

Validate that the selected backend has a matching config.

Returns a list of validation error messages. Empty list means valid.

`EventsModule`

source

CQRS + Event Sourcing: CommandBusProtocol, QueryBusProtocol, EventBusProtocol, EventStoreProtocol, sagas.

Call configure to configure the events subsystem.

Usage

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

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

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

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

configure

def configure(
    cls,
    config: Any | None = None,
    handler_modules: list[str] | None = None
) -> DynamicModule

source

Create an EventsModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	Any \| None	EventsConfig or ``None`` for framework defaults (in-memory event store).
`handler_modules`	list[str] \| None	Python module paths to auto-discover event/command handlers from (e.g. ``["myapp.handlers"]``).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

scope

def scope(
    cls,
    *handlers: type,
    **kwargs: Any
) -> DynamicModule

source

Scope event/command/query handler classes into a feature module.

Registers the given handler classes as providers so they are resolved by the bus implementations. The parent module graph must already include EventsModule.configure — this does not create a new event store or bus.

Uses the anonymous token pattern so both configure() and scope() can coexist in the same compiled graph without a ModuleDuplicateError.

Example

@module(
    imports=[
        EventsModule.configure(config),
        EventsModule.scope(
            CreateOrderHandler,
            GetOrderHandler,
            OrderShippedHandler,
        ),
    ]
)
class OrderFeatureModule(Module):
    pass

@module(
    imports=[
        EventsModule.configure(config),
        EventsModule.scope(
            CreateOrderHandler,
            GetOrderHandler,
            OrderShippedHandler,
        ),
    ]
)
class OrderFeatureModule(Module):
    pass

Parameters

Parameter	Type	Description

Returns

Type	Description
DynamicModule	A DynamicModule scoped to this feature.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return an in-memory EventsModule for unit testing.

Registers an in-memory event store with no external broker connections. Suitable for testing event handlers and CQRS logic.

Returns

Type	Description
DynamicModule	A DynamicModule backed by in-memory event storage.

`EventsProvider`

source

Thin orchestrator that coordinates Events sub-providers (M-01).

Delegates all setup, lifecycle, and container-registration work to:

StoreSubProvider
BusSubProvider
HandlerSubProvider
ManagerSubProvider

Public API is unchanged from the monolithic version.

__init__

def __init__(
    config: EventsConfig | dict[str, Any] | None = None,
    handler_modules: list[str] | None = None
) -> None

source

from_config

def from_config(
    cls,
    config: EventsConfig,
    **context: Any
) -> EventsProvider

source

Create EventsProvider from config.

config

property config() -> EventsConfig | None

source

config

def config(value: EventsConfig) -> None

source

Allow orchestrator to auto-inject config from QuadkitConfig.

event_store

property event_store() -> Any

source

snapshot_manager

property snapshot_manager() -> Any

source

command_bus

property command_bus() -> Any

source

query_bus

property query_bus() -> Any

source

event_bus

property event_bus() -> Any

source

projection_manager

property projection_manager() -> Any

source

saga_manager

property saga_manager() -> Any

source

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Start buses, managers, and wire optional tracer.

shutdown

async def shutdown() -> None

source

Shutdown all components.

Buses first: an event accepted by publish() is a promise, and the drain tasks that keep it are stopped only after the queued events have been handed to their handlers. Stores and managers follow, so a handler that still needs them finds them up.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Aggregate health across stores and buses.

`HandlerInfo`

source

Information about a registered handler.

`HandlerRegistry`

source

Registry for handler discovery and registration.

The registry provides:

Auto-discovery of handlers from modules
Registration with command, query, and event buses
Tracking of handler-message relationships

Example

# Create registry
registry = HandlerRegistry()

# Discover handlers from modules
registry.discover("app.handlers")

# Register with buses
registry.register_with_buses(
    command_bus=command_bus,
    query_bus=query_bus,
    event_bus=event_bus
)

# Or use DI container for handler instantiation
registry.register_with_container(container)

__init__

def __init__() -> None

source

Initialize the handler registry.

load_decorators

def load_decorators() -> None

source

Load handlers from the global decorator registry (M-17).

Call this explicitly to register handlers that were annotated with @command_handler, @event_handler, etc.

register_command_handler

def register_command_handler(
    command_type: type,
    handler_type: Any
) -> None

source

Parameters

Parameter	Type	Description
`command_type`	type	The command class
`handler_type`	Any	The handler class

register_query_handler

def register_query_handler(
    query_type: type,
    handler_type: Any
) -> None

source

Parameters

Parameter	Type	Description
`query_type`	type	The query class
`handler_type`	Any	The handler class

register_event_handler

def register_event_handler(
    event_type: type,
    handler_type: Any
) -> None

source

Parameters

Parameter	Type	Description
`event_type`	type	The event class
`handler_type`	Any	The handler class

discover

def discover(
    module_path: str,
    recursive: bool = True
) -> int

source

Discover handlers from a module or package.

Scans the module for classes that inherit from handler base classes and registers them automatically.

Parameters

Parameter	Type	Description
`module_path`	str	Python module path (e.g., "app.handlers")
`recursive`	bool	Whether to scan submodules

Returns

Type	Description
int	Number of handlers discovered

register_with_buses

def register_with_buses(
    command_bus: CommandBusProtocol | None = None,
    query_bus: QueryBusProtocol | None = None,
    event_bus: EventBusProtocol | None = None,
    handler_factory: Callable[[type], Any] | None = None
) -> None

source

Parameters

Parameter	Type	Description
`command_bus`	CommandBusProtocol \| None	Command bus to register with
`query_bus`	QueryBusProtocol \| None	Query bus to register with
`event_bus`	EventBusProtocol \| None	Event bus to register with
`handler_factory`	Callable[[type], Any] \| None	Factory function to instantiate handlers

get_command_handlers

def get_command_handlers() -> dict[type, type[CommandHandlerProtocol[Any, Any]]]

source

Get all registered command handlers.

get_query_handlers

def get_query_handlers() -> dict[type, type[QueryHandlerProtocol[Any, Any]]]

source

Get all registered query handlers.

get_event_handlers

def get_event_handlers() -> dict[type, list[type[EventHandlerProtocol[Any]]]]

source

Get all registered event handlers.

`IdempotentCommand`

source

Idempotent command extended with aggregate-targeting helpers.

`InMemoryEventStore`

source

In-memory event store for testing and development.

M-19: Uses a global log and type index for O(n) and O(matching) streaming performance.

Example

store = InMemoryEventStore()

# Save events
await store.append(
    stream_id="order-123",
    events=[OrderCreated(...)],
    expected_version=0
)

# Load events
events = await store.read("order-123")

store = InMemoryEventStore()

# Save events
await store.append(
    stream_id="order-123",
    events=[OrderCreated(...)],
    expected_version=0
)

# Load events
events = await store.read("order-123")

__init__

def __init__(
    schema_evolution: SchemaEvolution | None = None,
    max_events_per_stream: int = 10000
) -> None

source

Initialize the in-memory store.

Parameters

Parameter	Type	Description
`schema_evolution`	SchemaEvolution \| None	Optional schema evolution handler.
`max_events_per_stream`	int	Maximum events retained per stream. Oldest events are evicted when the limit is exceeded. Defaults to 10,000 — set to 0 for unbounded.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Report readiness for the process-local in-memory store.

The in-memory backend has no external dependency to probe. Reporting it explicitly avoids making a configured offline application look partially unknown to the EventsProvider health aggregate.

append

async def append(
    stream_id: str,
    events: list[Event],
    expected_version: int | None = None
) -> int

source

Append events to an in-memory stream with optimistic concurrency.

Returns

Type	Description
int	New stream version.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Event]

source

Read events from a stream (M-15).

get_stream_version

async def get_stream_version(stream_id: str) -> int

source

Get the current version of a stream.

stream_exists

async def stream_exists(stream_id: str) -> bool

source

Check the logical version, even after full compaction.

get_stream_info

async def get_stream_info(stream_id: str) -> StreamInfo | None

source

Report the logical version and metadata of the retained suffix.

get_stream_ids

async def get_stream_ids() -> list[str]

source

Get all stream IDs (for testing/debugging).

compact

async def compact(
    stream_id: str,
    up_to_version: int
) -> int

source

Purge events for a stream up to (and including) the given version (MF-04).

stream_all

async def stream_all(
    from_position: int = 0,
    batch_size: int = 100,
    partition: int | None = None,
    total_partitions: int | None = None
) -> AsyncIterator[Event]

source

Stream all events across all streams.

M-19: O(n) from position using _global_log — no merging required.

stream_by_type

async def stream_by_type(
    event_types: list[str],
    from_position: int = 0
) -> AsyncIterator[Event]

source

Stream events filtered by type.

M-19: O(matching_events) not O(total_events) using type index.

get_all_streams

def get_all_streams() -> list[str]

source

Get all stream IDs (for testing/debugging).

get_global_position

def get_global_position() -> int

source

Get current global position.

clear

def clear() -> None

source

Clear all events (for testing).

async def close() -> None

source

Close the in-memory event store (no-op).

`InMemoryEventStoreConfig`

source

Configuration for in-memory event store.

`InMemorySnapshotStore`

source

In-memory snapshot store for testing and development.

M-14: Uses Snapshot objects throughout (save_snapshot / get_latest / get_by_version). The legacy tuple-based save/load methods are provided by the AbstractSnapshotStore base class via backward-compat shims.

Example

from quadkit.events.types import Snapshot
store = InMemorySnapshotStore()

# Save snapshot
await store.save_snapshot(Snapshot(
    aggregate_id="order-123",
    aggregate_type="Order",
    version=100,
    state={"status": "completed", "total": 150.00}
))

# Load latest snapshot
snapshot = await store.get_latest("order-123")

from quadkit.events.types import Snapshot
store = InMemorySnapshotStore()

# Save snapshot
await store.save_snapshot(Snapshot(
    aggregate_id="order-123",
    aggregate_type="Order",
    version=100,
    state={"status": "completed", "total": 150.00}
))

# Load latest snapshot
snapshot = await store.get_latest("order-123")

__init__

def __init__(max_snapshots_per_aggregate: int = 5) -> None

source

Initialize the in-memory snapshot store.

Parameters

Parameter	Type	Description
`max_snapshots_per_aggregate`	int	Maximum snapshots to keep per aggregate

save_snapshot

async def save_snapshot(snapshot: Snapshot) -> None

source

Save a snapshot (M-14: accepts Snapshot object).

get_latest

async def get_latest(aggregate_id: str) -> Snapshot | None

source

Load the latest snapshot for an aggregate.

get_by_version

async def get_by_version(
    aggregate_id: str | UUID,
    version: int
) -> Snapshot | None

source

Get the snapshot for a specific version.

delete_old_snapshots

async def delete_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int = 3
) -> int

source

Delete old snapshots, keeping only the most recent ones.

clear

def clear() -> None

source

Clear all snapshots (for testing).

async def close() -> None

source

Close the in-memory snapshot store (no-op).

`IntegrationEvent`

source

Event for cross-service communication.

`KnownEventSetRegistry`

source

Tracks the set of (event_type, schema_version) pairs a consumer handles.

Each consumer gets its own registry, keyed by consumer_id.

__init__

def __init__(consumer_id: str) -> None

source

consumer_id

property consumer_id() -> str

source

all_known

property all_known() -> frozenset[EventTypeVersion]

source

def register(types: list[EventTypeVersion]) -> None

source

is_known

def is_known(
    event_type: str,
    version: int
) -> bool

source

has_type

def has_type(event_type: str) -> bool

source

`Message`

source

Base class for all messages in the Events system.

with_metadata

def with_metadata(metadata: MessageMetadata) -> Message

source

Create a copy of the message with new metadata.

with_correlation_id

def with_correlation_id(correlation_id: UUID) -> Message

source

Create a copy with a correlation ID.

with_causation_id

def with_causation_id(causation_id: UUID) -> Message

source

Create a copy with a causation ID.

correlation_id

property correlation_id() -> UUID | None

source

Get correlation ID from metadata.

causation_id

property causation_id() -> UUID | None

source

Get causation ID from metadata.

to_dict

def to_dict() -> dict[str, Any]

source

Convert message to dictionary.

from_dict

def from_dict(
    cls,
    data: dict[str, Any]
) -> Message

source

Create message from dictionary.

`MessageMetadata`

source

Metadata attached to messages for tracking and correlation.

with_correlation

def with_correlation(correlation_id: UUID) -> MessageMetadata

source

Create new metadata with correlation ID.

with_causation

def with_causation(causation_id: UUID) -> MessageMetadata

source

Create new metadata with causation ID.

with_user

def with_user(user_id: str) -> MessageMetadata

source

Create new metadata with user ID.

with_tenant

def with_tenant(tenant_id: str) -> MessageMetadata

source

Create new metadata with tenant ID.

with_custom

def with_custom(
    key: str,
    value: Any
) -> MessageMetadata

source

Create new metadata with additional custom field.

`MessageType`

source

Types of messages in the event-driven system.

`MiddlewareInfo`

source

Information about registered middleware.

`MongoDBConfig`

source

MongoDB connection configuration for store implementations.

Used directly by MongoDBSnapshotStore for motor connection lifecycle management.

`MongoDBEventStore`

source

MongoDB event store implementation using ``DocumentStoreProtocol``.

Uses the a document-store extension collection abstraction for standard CRUD while keeping direct motor access for change streams and transactions.

Features: - Optimistic concurrency with version checking - Global event ordering with sequence counters - Efficient event streaming with cursors - Automatic index creation - Change streams support for real-time events

Example

store = MongoDBEventStore(document_store, config)
await store.connect()

try:
    await store.save("order-123", [event], expected_version=0)
finally:
    await store.close()

__init__

def __init__(
    document_store: DocumentStoreProtocol,
    config: Any,
    event_serializer: Any | None = None
) -> None

source

Initialize MongoDB event store.

Parameters

Parameter	Type	Description
`document_store`	DocumentStoreProtocol	A connected ``DocumentStoreProtocol`` (e.g. MongoDBDocumentStore).
`config`	Any	Event store configuration with collection names.
`event_serializer`	Any \| None	Optional custom event serializer.

events

property events() -> CollectionProtocol

source

Events collection.

counters

property counters() -> CollectionProtocol

source

Counters collection.

connect

async def connect() -> None

source

Connect to MongoDB and set up collections.

append

async def append(
    stream_id: str,
    events: list[Event],
    expected_version: int | None = None
) -> int

source

Append events to a stream with optimistic concurrency.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.
`events`	list[Event]	Events to save.
`expected_version`	int \| None	Expected current version.

Returns

Type	Description
int	New stream version.

Raises

Exception	Description
ConcurrencyError	If version conflict occurs.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Event]

source

Read events from a stream.

get_stream_version

async def get_stream_version(stream_id: str) -> int

source

Get current stream version.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to check.

Returns

Type	Description
int	Current version (0 if stream doesn't exist).

stream_all

async def stream_all(
    from_position: int = 0,
    batch_size: int = 100,
    partition: int | None = None,
    total_partitions: int | None = None
) -> AsyncIterator[Event]

source

Stream all events from a global position.

stream_by_type

async def stream_by_type(
    event_types: list[str],
    from_position: int = 0
) -> AsyncIterator[Event]

source

Stream events filtered by type.

watch_events

async def watch_events(event_types: list[str] | None = None) -> AsyncIterator[Event]

source

Watch for new events using MongoDB change streams.

Note: This requires direct motor access for change stream support. Falls back to polling if the underlying store doesn’t support watch().

Parameters

Parameter	Type	Description
`event_types`	list[str] \| None	Optional list of event types to filter.

Yields

Type	Description
AsyncIterator[Event]	New events as they are inserted.

find_by_type

async def find_by_type(
    event_type: str,
    limit: int = 100,
    offset: int = 0
) -> list[Event]

source

Get events of a specific type.

Parameters

Parameter	Type	Description
`event_type`	str	Event type name.
`limit`	int	Maximum events to return.
`offset`	int	Number of events to skip.

Returns

Type	Description
list[Event]	List of events.

get_events_count

async def get_events_count(stream_id: str | None = None) -> int

source

Get total event count.

Parameters

Parameter	Type	Description
`stream_id`	str \| None	Optional stream to count (None for all).

Returns

Type	Description
int	Event count.

delete_stream

async def delete_stream(stream_id: str) -> int

source

Delete all events for a stream.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to delete.

Returns

Type	Description
int	Number of deleted events.

compact

async def compact(
    stream_id: str,
    up_to_version: int
) -> int

source

Purge events for a stream up to (and including) the given version (MF-04).

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to compact.
`up_to_version`	int	Maximum version to delete.

Returns

Type	Description
int	Number of events purged.

async def close() -> None

source

Close the MongoDB connection.

`MongoDBEventStoreConfig`

source

Configuration for MongoDB event store (top-level user config).

Used by EventsConfig.mongodb <quadkit.events.config.EventsConfig.mongodb>. For the store-level connection config used by MongoDBSnapshotStore, see quadkit.events.stores.mongodb.MongoDBConfig.

validate_connection_string

def validate_connection_string(
    cls,
    v: SecretStr | str
) -> SecretStr

source

Validate and coerce MongoDB connection string to SecretStr.

`MongoDBSnapshotStore`

source

MongoDB snapshot store implementation.

This store manages aggregate snapshots for optimized loading.

Example

config = MongoDBConfig(uri="mongodb://localhost:27017")
store = MongoDBSnapshotStore(config)
await store.connect()

# Save snapshot
await store.save(Snapshot(
    aggregate_id="order-123",
    aggregate_type="Order",
    version=100,
    state=order.to_dict(),
))

# Load latest
snapshot = await store.get_latest("order-123")

__init__

def __init__(config: MongoDBConfig) -> None

source

Initialize MongoDB snapshot store.

Parameters

Parameter	Type	Description
`config`	MongoDBConfig	MongoDB configuration.

snapshots

property snapshots() -> AsyncIOMotorCollection

source

connect

async def connect() -> None

source

Connect to MongoDB.

save_snapshot

async def save_snapshot(snapshot: Snapshot) -> None

source

Save a snapshot (M-14 standardized API).

get_latest

async def get_latest(aggregate_id: str) -> Snapshot | None

source

Get latest snapshot (M-14 standardized API).

get_by_version

async def get_by_version(
    aggregate_id: str | UUID,
    version: int
) -> Snapshot | None

source

Get a specific snapshot version (M-14).

delete_old_snapshots

async def delete_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int = 3
) -> int

source

Delete old snapshots, keeping only the most recent ones.

Parameters

Parameter	Type	Description
`aggregate_id`	str \| UUID	Aggregate identifier.
`keep_count`	int	Number of snapshots to keep.

Returns

Type	Description
int	Number of deleted snapshots.

async def close() -> None

source

Close the MongoDB connection.

`PagedResult`

source

Container for paginated results.

Attributes: items: The items in the current page. total: Total number of items across all pages. page: Current page number. page_size: Number of items per page.

Example

result = PagedResult(
    items=[...],
    total=150,
    page=2,
    page_size=20
)
logger.info(result.total_pages)  # 8
logger.info(result.has_next)  # True

__init__

def __init__(
    items: list[TResult],
    total: int,
    page: int,
    page_size: int
) -> None

source

Initialize paged result.

Parameters

Parameter	Type	Description
`items`	list[TResult]	Items in the current page.
`total`	int	Total item count.
`page`	int	Current page number.
`page_size`	int	Items per page.

total_pages

property total_pages() -> int

source

Get total number of pages.

has_next

property has_next() -> bool

source

Check if there is a next page.

Returns

Type	Description
bool	True if there are more pages after current.

has_prev

property has_prev() -> bool

source

Check if there is a previous page.

Returns

Type	Description
bool	True if there are pages before current.

to_dict

def to_dict() -> dict[str, Any]

source

Convert to dictionary.

Returns

Type	Description
dict[str, Any]	Dictionary representation of paged result.

`PaginatedQuery`

source

Query with built-in pagination support.

offset

property offset() -> int

source

Get query offset based on page.

limit

property limit() -> int

source

Get query limit.

`PostgresEventStoreConfig`

source

PostgreSQL connection configuration.

Attributes: dsn: Database connection string (treated as a secret). pool_min_size: Minimum connection pool size. pool_max_size: Maximum connection pool size. command_timeout: Command timeout in seconds. events_table: Name of the events table. snapshots_table: Name of the snapshots table. auto_create_tables: Whether to create tables automatically.

validate_dsn

def validate_dsn(
    cls,
    v: SecretStr | str | None
) -> SecretStr | None

source

Validate and normalize the DSN. Returns None when not provided.

`ProjectionState`

source

States of a projection.

`ProjectionUpdatedEvent`

source

Emitted when a read-model projection is rebuilt or updated.

Attributes: projection_name: Canonical name of the projection that was updated. events_processed: Number of domain events processed in this update.

`Query`

source

Represents a request for data.

`QueryBusConfig`

source

Configuration for query bus.

`QueryBusImpl`

source

Query bus for dispatching queries.

Handlers are registered explicitly via register() or through HandlerRegistry.register_with_buses() during provider registration.

Example

# Queries are automatically routed to handlers
orders = await bus.execute(GetOrdersByCustomerQuery(
    customer_id="cust-123",
    status="pending"
))

# Queries are automatically routed to handlers
orders = await bus.execute(GetOrdersByCustomerQuery(
    customer_id="cust-123",
    status="pending"
))

__init__

def __init__(
    middlewares: list[Any] | None = None,
    config: Any | None = None
) -> None

source

Initialize the query bus.

execute

async def execute(query: Query[TResult]) -> TResult

source

Execute a query.

Parameters

Parameter	Type	Description
`query`	Query[TResult]	The query to execute.

Returns

Type	Description
TResult	Query result.

Raises

Exception	Description
HandlerNotFoundError	If no handler is registered.
QueryExecutionError	If query execution fails.

`QueryHandlerProtocol`

source

Base class for query handlers.

Query handlers process queries and return results. Each query type should have exactly one handler. Queries should not modify state.

Example

class GetOrdersHandler(QueryHandlerProtocol[GetOrdersQuery, list[OrderDTO]]):
    def __init__(self, read_model: OrderReadModel):
        self.read_model = read_model

    async def handle(self, query: GetOrdersQuery) -> list[OrderDTO]:
        return await self.read_model.get_orders(
            customer_id=query.customer_id,
            status=query.status
        )

handle

async def handle(query: TQuery) -> TResult

source

Handle a query.

Parameters

Parameter	Type	Description
`query`	TQuery	The query to handle

Returns

Type	Description
TResult	Query result

`QueryResult`

source

Result of query execution.

`RedisEventStore`

source

Redis-backed AbstractEventStore for lightweight deployments.

Uses the StateStoreProtocol protocol for storage, so no direct Redis dependency is introduced into this package. Suitable for single-node or low-throughput deployments where a full relational or document store is not required.

Key layout

{namespace}:{stream_id}:events  — JSON array of serialised event dicts
{namespace}:{stream_id}:version — current integer version for the stream
{namespace}:_streams            — JSON array of all known stream IDs

{namespace}:{stream_id}:events  — JSON array of serialised event dicts
{namespace}:{stream_id}:version — current integer version for the stream
{namespace}:_streams            — JSON array of all known stream IDs

Example

from quadkit.events.stores.redis import RedisEventStore

store = RedisEventStore(state_store, namespace="myapp:events")
await store.append("order-123", [OrderCreated(...)])
events = await store.read("order-123")

from quadkit.events.stores.redis import RedisEventStore

store = RedisEventStore(state_store, namespace="myapp:events")
await store.append("order-123", [OrderCreated(...)])
events = await store.read("order-123")

__init__

def __init__(
    store: StateStoreProtocol,
    namespace: str = 'quadkit:events'
) -> None

source

Initialise the Redis-backed event store.

Parameters

Parameter	Type	Description
`store`	StateStoreProtocol	A StateStoreProtocol instance (e.g. a Redis-backed implementation).
`namespace`	str	Key prefix used for all stored keys.

append

async def append(
    stream_id: str,
    events: list[Event],
    expected_version: int | None = None
) -> int

source

Append events to stream_id with optimistic concurrency control.

Parameters

Parameter	Type	Description
`stream_id`	str	Unique stream identifier.
`events`	list[Event]	Events to append. Empty list is a no-op.
`expected_version`	int \| None	If provided, the current stream version must equal this value or ConcurrencyError is raised.

Returns

Type	Description
int	New stream version after the append.

Raises

Exception	Description
ConcurrencyError	If expected_version does not match the actual current version of the stream.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Event]

source

Read events from stream_id.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to read from.
`from_version`	int	Lowest version to include (inclusive, default 0 = all).
`to_version`	int \| None	Highest version to include (inclusive).
`limit`	int \| None	Maximum number of events to return.

Returns

Type	Description
list[Event]	Ordered list of Event objects.

load

async def load(
    stream_id: str,
    after_version: int = 0
) -> list[Event]

source

Load events after after_version (non-inclusive lower bound).

This alias is used by several base-class helpers.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to load events from.
`after_version`	int	Events with version <= this value are excluded.

Returns

Type	Description
list[Event]	Ordered list of events.

get_stream_version

async def get_stream_version(stream_id: str) -> int

source

Return the current version of stream_id (0 if the stream is empty).

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.

Returns

Type	Description
int	Current integer version.

get_version

async def get_version(stream_id: str) -> int

source

Public alias for get_stream_version.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream identifier.

Returns

Type	Description
int	Current integer version.

stream_all

async def stream_all(
    from_position: int = 0,
    batch_size: int = 100,
    partition: int | None = None,
    total_partitions: int | None = None
) -> AsyncIterator[Event]

source

Stream all events across all known streams.

Note: Cross-stream global ordering is not guaranteed by this implementation. Streams are iterated in registration order and events within each stream are returned in version order. For strict global ordering, use the PostgreSQL or MongoDB stores.

Parameters

Parameter	Type	Description
`from_position`	int	Events with ``sequence_number`` <= this value are skipped.
`batch_size`	int	Unused; present for interface compatibility.
`partition`	int \| None	Optional partition index for sharding.
`total_partitions`	int \| None	Total number of partitions.

Yields

Type	Description
AsyncIterator[Event]	Events in stream-registration order, version order within each stream.

compact

async def compact(
    stream_id: str,
    up_to_version: int
) -> int

source

Purge events for stream_id up to and including up_to_version.

Parameters

Parameter	Type	Description
`stream_id`	str	Stream to compact.
`up_to_version`	int	Events at or below this version are removed.

Returns

Type	Description
int	Number of events deleted.

async def close() -> None

source

Close the event store.

No-op for this implementation; the underlying StateStoreProtocol manages its own lifecycle.

`Snapshot`

source

Aggregate state snapshot for performance optimization.

`SnapshotConfig`

source

Configuration for snapshotting behavior.

`SnapshotManager`

source

Manages snapshot creation and retrieval for aggregates.

Snapshots avoid reapplying earlier events; loading still depends on the retained suffix and the event store’s version/read implementation.

Example

manager = SnapshotManager(
    event_store=event_store,
    snapshot_store=snapshot_store,
    policy=EventCountPolicy(100)
)

# Load aggregate with snapshot acceleration
state, events = await manager.load_with_snapshot(
    aggregate_id="order-123",
    aggregate_type="Order"
)

# Auto-snapshot after save if policy triggers
await manager.save_and_maybe_snapshot(
    aggregate_id="order-123",
    aggregate_type="Order",
    events=[...],
    current_state=state,
    expected_version=100
)

__init__

def __init__(
    event_store: EventStoreProtocol,
    snapshot_store: SnapshotStoreProtocol,
    policy: SnapshotPolicy | None = None,
    config: SnapshotConfig | None = None
)

source

Initialize the snapshot manager.

Parameters

Parameter	Type	Description
`event_store`	EventStoreProtocol	The event store instance
`snapshot_store`	SnapshotStoreProtocol	The snapshot store instance
`policy`	SnapshotPolicy \| None	Snapshot creation policy (default: EventCountPolicy(100))
`config`	SnapshotConfig \| None	Snapshot configuration

load_with_snapshot

async def load_with_snapshot(
    aggregate_id: str | UUID,
    aggregate_type: str
) -> tuple[dict[str, Any] | None, list[Any], int]

source

Load aggregate state using snapshot acceleration.

Load the latest snapshot and its remaining events, checking that the stream version is stable and the returned suffix has the expected size. The checks do not establish snapshot truth or transactional isolation.

Returns

Type	Description
tuple[dict[str, Any] \| None, list[Any], int]	Tuple of (snapshot_state, events_to_replay, starting_version)

save_and_maybe_snapshot

async def save_and_maybe_snapshot(
    aggregate_id: str | UUID,
    aggregate_type: str,
    events: list[Any],
    current_state: dict[str, Any],
    expected_version: int
) -> bool

source

Save events and create snapshot if policy triggers.

Returns

Type	Description
bool	True if a snapshot was created

create_snapshot

async def create_snapshot(
    aggregate_id: str | UUID,
    aggregate_type: str,
    version: int,
    state: dict[str, Any]
) -> Snapshot

source

Create and store a snapshot.

force_snapshot

async def force_snapshot(
    aggregate_id: str | UUID,
    aggregate_type: str,
    state_extractor: Callable[[], dict[str, Any]],
    *,
    expected_version: int | None = None
) -> Snapshot

source

Snapshot caller-supplied state at the store’s current version.

The caller must ensure that the state actually represents that version. This operation is not atomic with event writes or state extraction.

cleanup_old_snapshots

async def cleanup_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int | None = None
) -> int

source

Clean up old snapshots for an aggregate.

`SnapshotStrategy`

source

Snapshot creation strategies.

`SqliteConfig`

source

SQLite configuration for event stores.

Table-name and DDL fields mirror PostgresEventStoreConfig so both SQL backends behave identically under default configuration.

`SqliteEventStore`

source

SQLite event store implementation.

This store uses the generic DatabaseProviderProtocol for async database access. Suitable for development, testing, and single-instance deployments.

__init__

def __init__(
    config: SqliteConfig | None = None,
    provider: DatabaseProviderProtocol | None = None,
    event_serializer: Any | None = None
) -> None

source

Initialize SQLite event store.

Parameters

Parameter	Type	Description
`config`	SqliteConfig \| None	SQLite configuration.
`provider`	DatabaseProviderProtocol \| None	The generic database provider injected via DI.
`event_serializer`	Any \| None	Optional custom event serializer.

register_event_type

def register_event_type(
    event_type: str,
    event_class: type
) -> None

source

connect

async def connect() -> None

source

Connection is managed by the provider. Tables are created if configured.

append

async def append(
    stream_id: str,
    events: list[Event],
    expected_version: int | None = None
) -> int

source

Append events to a stream with optimistic concurrency.

read

async def read(
    stream_id: str,
    from_version: int = 0,
    to_version: int | None = None,
    limit: int | None = None
) -> list[Event]

source

Read events from a stream (M-15).

get_stream_version

async def get_stream_version(stream_id: str) -> int

source

Get the stream high-water version, including compacted events.

stream_exists

async def stream_exists(stream_id: str) -> bool

source

Check the logical version, including a compacted stream’s marker.

get_stream_info

async def get_stream_info(stream_id: str) -> StreamInfo | None

source

Report logical version and metadata of currently retained events.

stream_all

async def stream_all(
    from_position: int = 0,
    batch_size: int = 100,
    partition: int | None = None,
    total_partitions: int | None = None
) -> AsyncIterator[Event]

source

Stream events after a global position, optionally by stream partition.

Partitioning scans ordered batches and hashes the stream ID in Python; SQLite does not provide the old query’s STR_HASH function. It uses the same MD5-prefix rule as the memory and Redis stores, but does not imply partition compatibility with every backend or a custom SQLite UDF.

stream_by_type

async def stream_by_type(
    event_types: list[str],
    from_position: int = 0
) -> AsyncIterator[Event]

source

Stream events filtered by type.

find_by_type

async def find_by_type(
    event_type: str,
    count: int = 100,
    start: int = 0
) -> list[Event]

source

Get events of a specific type.

get_events_count

async def get_events_count(stream_id: str | None = None) -> int

source

Get total event count.

delete_stream

async def delete_stream(stream_id: str) -> int

source

Delete all events for a stream.

get_stream_ids

async def get_stream_ids() -> list[str]

source

Get all stream IDs in the store.

compact

async def compact(
    stream_id: str,
    up_to_version: int
) -> int

source

Purge events for a stream up to (and including) the given version (MF-04).

async def close() -> None

source

Close the database connection.

`SqliteSnapshotStore`

source

SQLite snapshot store implementation.

This store manages aggregate snapshots using a generic DatabaseProviderProtocol.

__init__

def __init__(
    config: SqliteConfig | None = None,
    provider: DatabaseProviderProtocol | None = None
) -> None

source

Initialize SQLite snapshot store.

Parameters

Parameter	Type	Description
`config`	SqliteConfig \| None	SQLite configuration.
`provider`	DatabaseProviderProtocol \| None	The generic database provider injected via DI.

connect

async def connect() -> None

source

Connection is managed by the provider.

save_snapshot

async def save_snapshot(snapshot: Snapshot) -> None

source

Save a snapshot (M-14 standardized API).

get_latest

async def get_latest(aggregate_id: str) -> Snapshot | None

source

Get latest snapshot (M-14 standardized API).

get_by_version

async def get_by_version(
    aggregate_id: str | UUID,
    version: int
) -> Snapshot | None

source

Get a specific snapshot version (M-14).

delete_old_snapshots

async def delete_old_snapshots(
    aggregate_id: str | UUID,
    keep_count: int = 3
) -> int

source

Delete old snapshots, keeping only the most recent ones.

async def close() -> None

source

Close the database connection.

`StreamInfo`

source

Information about an event stream.

`StreamPosition`

source

Starting position for event streams.

`ValueObject`

source

Base class for value objects.

Value objects:

Are immutable (frozen=True)
Have no identity
Are compared by value (all fields)
Can be freely shared

`VersionAwareSubscription`

source

Wraps an EventBus subscription with type+version verification.

Decision tree on incoming event:

Is (event_type, version) in known_set? -> invoke handler.
Is the type known but version unknown? -> try evolution.get_migration_path(). If path exists, migrate the data and invoke handler. If no path, emit EventSchemaVersionSkew alert.
Is the type completely unknown? -> emit UnknownEventTypeReceived alert.

__init__

def __init__(
    consumer_id: str,
    bus: EventBusProtocol,
    known_set: KnownEventSetRegistry,
    evolution: SchemaEvolution | None,
    alert_dispatcher: AlertDispatcherProtocol
) -> None

source

async def subscribe(
    event_type_pattern: str,
    handler: Callable[[Any], Awaitable[None]]
) -> None

source

`WebhookDispatcher`

source

Delivers domain events to registered external webhook endpoints.

Signs each outbound POST with a cryptographic signature via the injected WebhookSignatureVerifierProtocol so that receivers can verify authenticity. Retries transient 5xx failures according to retry_policy. Delivery failures for one endpoint are logged and do not prevent fan-out to remaining endpoints.

Example

dispatcher = WebhookDispatcher(
    http_client=http_client,
    signature_verifier=HMACWebhookVerifier(),
    retry_policy=RetryConfig(max_attempts=3),
)
await dispatcher.dispatch(user_created_event, registered_endpoints)

dispatcher = WebhookDispatcher(
    http_client=http_client,
    signature_verifier=HMACWebhookVerifier(),
    retry_policy=RetryConfig(max_attempts=3),
)
await dispatcher.dispatch(user_created_event, registered_endpoints)

__init__

def __init__(
    http_client: HTTPClientProtocol,
    signature_verifier: WebhookSignatureVerifierProtocol,
    retry_policy: RetryConfig | None = None
) -> None

source

Initialise the dispatcher.

Parameters

Parameter	Type	Description
`http_client`	HTTPClientProtocol	HTTP client used for outbound POST requests.
`signature_verifier`	WebhookSignatureVerifierProtocol	Used to compute the outbound HMAC signature included in ``X-Webhook-Signature``.
`retry_policy`	RetryConfig \| None	Retry configuration; defaults to 3 attempts with exponential backoff when not provided.

dispatch

async def dispatch(
    event: DomainEvent,
    endpoints: list[WebhookEndpoint]
) -> None

source

Sign and POST event to each eligible registered endpoint.

Only endpoints whose event_types filter matches the event are delivered to. Each delivery is retried per retry_policy on transient (5xx) failures. Failures are logged but do not abort delivery to remaining endpoints.

Parameters

Parameter	Type	Description
`event`	DomainEvent	Domain event to fan-out.
`endpoints`	list[WebhookEndpoint]	Registered endpoints to consider for delivery.

`WebhookEndpoint`

source

An external endpoint registered to receive domain events.

Attributes: url: HTTP(S) URL to POST events to. secret: Shared secret used to compute the HMAC signature. event_types: Set of event type names this endpoint subscribes to. When None, all events are delivered.

accepts

def accepts(event: DomainEvent) -> bool

source

Return True when this endpoint should receive event.

Parameters

Parameter	Type	Description
`event`	DomainEvent	Domain event to test against the filter.

Returns

Type	Description
bool	``True` `when` `event_types` `is` `None` `(all events) or the event's type name is contained in` `event_types``.

Functions

`bind_event_handler`

bind_event_handler

async def bind_event_handler(
    handler: Callable[Ellipsis, Awaitable[Any]],
    container: Any,
    *,
    skip_if_missing_deps: bool = True
) -> Callable[[Any], Awaitable[Any]] | None

source

Bind extra DI-injected parameters on an async ``@event_handler`` function.

Parameters after the leading event argument are resolved from container using their type annotations:

Parameters typed Svc | None or carrying a default value receive None / their default when Svc is not registered.
Required parameters missing from container return None (skipping subscription) when skip_if_missing_deps=True, or raise when False.
Handlers taking only (event) are returned unchanged with zero wrapper allocation.

`clear_handler_registry`

clear_handler_registry

def clear_handler_registry() -> None

source

Clear all registered handlers from the global handler registry.

Intended for use in tests to reset handler state between test runs. Clears both the HandlerRegistry instances and the global decorator registry.

`command_handler`

command_handler

def command_handler(
    command_type: type[Command],
    *,
    name: str | None = None,
    **metadata: Any
) -> Callable[[Callable[P, R]], Callable[P, R]]

source

Decorator to register a function as a command handler.

Parameters

Parameter	Type	Description
`command_type`	type[Command]	The Command class this handler handles.
`name`	str \| None	Optional custom name for the handler. **metadata: Additional metadata to attach to the handler.

Returns

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Decorator function.

Example

@command_handler(CreateUserCommand)
async def handle_create_user(command: CreateUserCommand) -> str:
    user = await user_service.create(command.username, command.email)
    return user.user_id

`event_handler`

event_handler

def event_handler(
    event_type: type[DomainEvent],
    *,
    name: str | None = None,
    priority: int = 0,
    **metadata: Any
) -> Callable[[Callable[P, R]], Callable[P, R]]

source

Decorator to register a function as an event handler.

Parameters

Parameter	Type	Description
`event_type`	type[DomainEvent]	The Event class this handler handles.
`name`	str \| None	Optional custom name for the handler.
`priority`	int	Handler priority (higher = executed first). **metadata: Additional metadata to attach to the handler.

Returns

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Decorator function.

Example

@event_handler(UserCreatedEvent, priority=10)
async def send_welcome_email(event: UserCreatedEvent) -> None:
    await email_service.send_welcome(event.email)

`from_bus`

from_bus

def from_bus(
    dispatcher: StreamDispatcher,
    event_store: AbstractEventStore,
    from_position: int = 0,
    event_types: list[str] | None = None
) -> EventStream[Event]

source

Build a catchup-plus-live stream over a dispatcher.

Parameters

Parameter	Type	Description
`dispatcher`	StreamDispatcher	StreamDispatcher that also receives store writes.
`event_store`	AbstractEventStore	Store used for the catch-up phase.
`from_position`	int	Global position to replay from. Defaults to 0.
`event_types`	list[str] \| None	Optional event type name filter.

Returns

Type	Description
EventStream[Event]	A hot stream emitting historical events first, then live events.

Note

Cancelling the consumer unsubscribes from the dispatcher.

`from_store`

from_store

def from_store(
    store: AbstractEventStore,
    from_position: int = 0,
    event_types: list[str] | None = None
) -> EventStream[Event]

source

Build a cold replay stream over an event store.

Parameters

Parameter	Type	Description
`store`	AbstractEventStore	Any event store exposing ``stream_all` `/` `stream_by_type``.
`from_position`	int	Global position to replay from. Defaults to 0.
`event_types`	list[str] \| None	Optional event type name filter.

Returns

Type	Description
EventStream[Event]	A cold stream of stored events.

Example

async for event in from_store(store, from_position=42):
    process(event)

`known_events`

known_events

def known_events(consumer_id: str) -> Callable[[type], type]

source

Class decorator that registers a known event set.

Reads KNOWN: list[EventTypeVersion] from the decorated class.

Raises

Exception	Description
TypeError	If the class does not define ``KNOWN``.

`query_handler`

query_handler

def query_handler(
    query_type: type[Query],
    *,
    name: str | None = None,
    cacheable: bool = False,
    cache_ttl: int | None = None,
    **metadata: Any
) -> Callable[[Callable[P, R]], Callable[P, R]]

source

Decorator to register a function as a query handler.

Parameters

Parameter	Type	Description
`query_type`	type[Query]	The Query class this handler handles.
`name`	str \| None	Optional custom name for the handler.
`cacheable`	bool	Whether results can be cached.
`cache_ttl`	int \| None	Cache time-to-live in seconds. **metadata: Additional metadata to attach to the handler.

Returns

Type	Description
Callable[[Callable[P, R]], Callable[P, R]]	Decorator function.

Example

@query_handler(GetUserQuery, cacheable=True, cache_ttl=300)
async def handle_get_user(query: GetUserQuery) -> UserDTO:
    return await user_repository.get(query.user_id)

`retry_with_resilience`

retry_with_resilience

def retry_with_resilience(policy: Any) -> Op[Event, Event]

source

Bridge a resilience policy into the core ``retry`` operator.

Parameters

Parameter	Type	Description
`policy`	Any	A resilience retry policy exposing ``should_retry(error)``.

Returns

Type	Description
Op[Event, Event]	A core-compatible retry operator.

`subscribe_event_handlers`

subscribe_event_handlers

async def subscribe_event_handlers(
    event_bus: EventBusProtocol,
    container: Any,
    handlers: Sequence[Callable[Ellipsis, Awaitable[Any]]],
    *,
    skip_if_missing_deps: bool = True
) -> list[tuple[type, Callable[[Any], Awaitable[Any]]]]

source

Bind DI dependencies for *handlers* and subscribe them onto *event_bus*.

Returns the list of (event_type, subscribed_callable) pairs so callers can unsubscribe cleanly during provider shutdown.

Exceptions

`AggregateNotFoundError`

source

Raised when aggregate is not found.

`CommandExecutionError`

source

Command execution error.

__init__

def __init__(
    message: str = 'Command execution failed',
    command_type: str | None = None,
    error: str | None = None,
    **kwargs: Any
) -> None

source

`ConcurrencyError`

source

Concurrency/optimistic locking error.

Raised when event version conflicts indicate concurrent modifications.

__init__

def __init__(
    message: str = 'Concurrency error',
    expected_version: int | None = None,
    actual_version: int | None = None,
    **kwargs: Any
) -> None

source

`DuplicateHandlerError`

source

Duplicate handler registration.

Raised when attempting to register a handler for a type that already has one.

__init__

def __init__(
    message: str = 'Duplicate handler',
    message_type: str | None = None,
    **kwargs: Any
) -> None

source

`EventError`

source

Base events/messaging error.

__init__

def __init__(
    message: str = 'Events error',
    **kwargs: Any
) -> None

source

`EventLoadError`

source

Raised when event loading fails.

__init__

def __init__(
    message: str = 'Event load error',
    **kwargs: Any
) -> None

source

`EventPersistenceError`

source

Raised when event persistence fails.

`EventStoreConnectionError`

source

Raised when event store connection fails.

__init__

def __init__(
    message: str = 'Event store connection error',
    **kwargs: Any
) -> None

source

`EventStoreError`

source

Raised for event store errors.

__init__

def __init__(
    message: str = 'Event store error',
    **kwargs: Any
) -> None

source

`HandlerNotFoundError`

source

Event/command handler not found.

Raised when no handler is registered for a specific event or command type.

__init__

def __init__(
    message: str = 'Handler not found',
    handler_type: str | None = None,
    message_type: str | None = None,
    **kwargs: Any
) -> None

source

`ProjectionBuildError`

source

Raised when projection building fails.

`ProjectionNotFoundError`

source

Raised when a projection is not found.

`QueryExecutionError`

source

Query execution error.

__init__

def __init__(
    message: str = 'Query execution failed',
    query_type: str | None = None,
    error: str | None = None,
    **kwargs: Any
) -> None

source

`SchemaError`

source

Raised for schema-related errors.

`SecurityError`

source

Raised for security-related errors.

`StreamNotFoundError`

source

Raised when stream is not found.

Parameters mirror the helper used in tests: (stream_type, stream_id).

__init__

def __init__(
    stream_type: str,
    stream_id: str,
    message: str | None = None,
    **kwargs: Any
) -> None

source

`StreamingError`

source

Raised for streaming-related errors.

`WebhookDeliveryError`

source

Raised when an outbound webhook delivery fails after all retry attempts.

Carries the target url and the HTTP status code returned by the remote endpoint so callers can log structured failure context.

__init__

def __init__(
    url: str,
    status: int,
    message: str = 'Webhook delivery failed',
    **kwargs: Any
) -> None

source