Skip to content

aaa.events

Events

Event

Bases: BaseModel

Immutable, CloudEvents-compatible envelope. The event log is the source of truth.

caused 

caused(
    type: str, data: dict[str, Any] | None = None, **kw: Any
) -> Event

Derive a follow-up event preserving the causal chain and stream. The id is a pure function of (parent, type, data), so re-deriving after a crash is idempotent.

BlobRef

Bases: BaseModel

Claim-check pointer to large payloads held in a BlobStore.

SchemaRegistry 

SchemaRegistry()

Versioned payload schemas with upcasters that migrate old events on read.

validate 

validate(event: Event) -> Event

Return the event upcast to the latest version and validated; unknown refs pass.

cloudevents

CloudEvents 1.0 (structured JSON mode) <-> aaa Event. Dependency-free by design: the output is spec-conformant and interoperates with the official SDK / any broker.

Stores

EventStore

Bases: Protocol

append async 

append(
    stream: str,
    events: Sequence[Event],
    expected_seq: int = ANY_SEQ,
) -> list[Event]

Atomically append; expected_seq is the stream's last seq (-1 = empty stream). Raises ConcurrencyError on mismatch. IDEMPOTENT: an event whose id is already in the stream is not appended again; the existing stored event is returned with dup=True.

subscribe 

subscribe(
    stream: str, from_seq: int = 0
) -> AsyncIterator[Event]

Replay from from_seq, then tail live appends.

has async 

has(stream: str, event_id: str) -> bool

O(1) 'is this event id already in the stream?' (ids are unique per stream).

run_index async 

run_index() -> list[RunInfo]

Every run's started event and (if any) completed event, for crash recovery.

scrub async 

scrub(stream: str) -> int

Erase every event's payload in stream (see aaa.events.event.scrubbed), keep the envelope, and drop the stream's snapshots and any bus copies of its events. Idempotent; returns the number of events in the stream. Only meaningful for finished runs.

delete_stream async 

delete_stream(stream: str) -> int

Remove stream entirely (events, snapshots, bus copies). Returns events removed.

aclose async 

aclose() -> None

Release connections/files. A no-op for the in-memory store.

InMemoryEventStore 

InMemoryEventStore(outbox: Any = None)

SQLiteEventStore 

SQLiteEventStore(
    path: str | Path,
    *,
    synchronous: str = "NORMAL",
    poll_s: float = 0.2,
)

Durable EventStore on stdlib sqlite3 (no extra dependency).

  • One dedicated thread owns the connection, so the event loop never blocks on disk.
  • WAL + synchronous=NORMAL: crash-safe (no corruption); a power loss may drop the last few commits, never reorder or tear them. Use synchronous="FULL" for strict durability.
  • Append is one IMMEDIATE transaction: the OCC check and the inserts are atomic.
  • Append is idempotent on (stream, id), which is what makes crash recovery safe.
  • subscribe wakes instantly for same-process appends and polls for other processes.

PostgresEventStore 

PostgresEventStore(
    pool: Pool, dsn: str, partitions: int, outbox: bool
)

trim_outbox async 

trim_outbox(keep_last: int = 100000) -> int

Delete bus entries older than the newest keep_last per partition. Consumers that fall further behind than this lose the gap; size it to your worst-case outage.

RedisEventStore 

RedisEventStore(
    r: Redis, partitions: int, prefix: str, outbox: bool
)

trim_outbox async 

trim_outbox(keep_last: int = 100000) -> int

Trim each partition's bus stream to roughly the newest keep_last entries.

Projections and snapshots

Projection 

Projection(
    initial: Callable[[], S],
    fold: Callable[[S, Event], S],
    *,
    name: str = "",
    version: int = 1,
    state_type: Any = None,
    every: int = 64,
)

Bases: Generic[S]

State = fold over a stream's events. fold(state, event) may mutate and return state (the state object is private to the load). With a name and state_type, the state is snapshotted every every events and later loads cost O(tail), not O(history). Bump version whenever the fold logic or state shape changes: stale snapshots are ignored.

Loaded dataclass 

Loaded(
    state: S,
    seq: int,
    from_snapshot: int = -1,
    replayed: int = 0,
)

Bases: Generic[S]

StoredSnapshot

Bases: BaseModel

A projection's state as of seq (inclusive). Pure derived data: always safe to drop.

SnapshotCapable

Bases: Protocol

save_snapshot async 

save_snapshot(snap: StoredSnapshot) -> None

Must be monotonic: ignore a snapshot whose seq is not newer than the stored one.

Buses

EventBus

Bases: Protocol

What the Engine needs from a bus (InProcessBus, DistributedBus and ClientBus all conform).

abort async 

abort() -> None

Hard stop that models a crash: nothing drained, nothing released.

drain async 

drain() -> None

Wait until everything currently queued has been handled.

InProcessBus 

InProcessBus(
    queue_size: int = 1024,
    *,
    redeliver_base_s: float = 0.2,
    redeliver_max_s: float = 5.0,
)

Bounded-queue bus. Publishing blocks when a subscriber is full (backpressure). concurrency=1 preserves publish order per subscription. Handler exceptions are logged and isolated: one bad handler never kills the bus or other subscribers. Subscriptions sharing a group compete for events (one delivery per group).

abort async 

abort() -> None

Hard stop (crash simulation / emergency shutdown): cancel workers, drop queues.

drain async 

drain() -> None

Wait until all queued and in-flight events (and their follow-ups) are handled.

DistributedBus 

DistributedBus(
    backend: BusBackend,
    *,
    member: str | None = None,
    batch: int = 64,
    poll_s: float = 0.5,
    lease_ttl_s: float = 6.0,
    tick_s: float = 1.0,
    handover_timeout_s: float = 30.0,
    redeliver_base_s: float = 0.2,
    redeliver_max_s: float = 5.0,
)

EventBus over a BusBackend. publish is a no-op: the store already enqueued the event atomically with appending it (see PostgresEventStore / RedisEventStore / MemoryBusBackend).

execution_guard 

execution_guard() -> ExecutionGuard

Cross-process mutual exclusion for side-effecting work (same lease semantics as partitions).

publish async 

publish(event: Event) -> None

No-op by design (the store enqueues atomically with the append).

__aexit__ async 

__aexit__(*exc: object) -> None

Graceful: stop consuming, let in-flight handlers finish, commit cursors, release leases.

abort async 

abort() -> None

Crash: cancel everything, release NOTHING (leases/membership must expire).

drain async 

drain(timeout: float = 15.0) -> None

Wait until this process's consumers have handled everything currently in the log.

ClientBus

For processes that only SUBMIT and OBSERVE runs (an API server, a CLI): it delivers and consumes nothing, so the process never becomes a worker. Pair it with a shared store; the workers' DistributedBus picks the events up from the store's outbox.

BusBackend

Bases: Protocol

read async 

read(
    partition: int, after: int, limit: int, wait_s: float
) -> list[tuple[int, Event]]

Entries with id > after, oldest first; block up to wait_s for new ones.

tail async 

tail(partition: int) -> int

Highest entry id in the partition (0 if empty).

try_lease async 

try_lease(
    group: str, partition: int, member: str, ttl_s: float
) -> bool

Acquire the lease, or renew it if member already holds it.

abandon async 

abandon() -> None

Crash simulation hook: drop connections WITHOUT releasing anything.

MemoryBusBackend 

MemoryBusBackend(
    partitions: int = 8,
    clock: Callable[[], float] = time.monotonic,
)

In-process reference backend (shared object = shared 'cluster'). Used by tests and as the executable specification the Postgres/Redis backends must match.

forget 

forget(stream: str) -> None

Erase the payload of this stream's queued entries (positions are ids, so entries stay).

partition_of 

partition_of(stream: str, partitions: int) -> int

Stable across processes and Python versions (unlike hash()).

Durable acknowledgement

defer_ack 

defer_ack() -> Deferral

Call from inside a handler. See Deferral.

Deferral 

Deferral(active: bool = True)

Holds the acknowledgement of the event currently being handled.

A handler that hands its real work to a background task calls defer_ack() first and release()s when that work is finished (in a finally). Until then a durable bus will not commit its cursor past the event, so if the process dies the event is redelivered. Releasing twice is harmless. Buses without durable delivery return an inert Deferral.

Execution guards

guard

Execution guards: mutual exclusion for side-effecting work across workers.

A guard answers "may I run THIS call right now?". It is a lease with a heartbeat: acquire it, renew it while working, and treat a failed renewal as "I am no longer the only one running this". LocalGuard covers a single process; LeaseGuard reuses the bus backend's partition-lease primitive (Postgres advisory lock / Redis SET NX PX / memory) so it works across processes with exactly the same failure semantics as partition ownership.

ExecutionGuard

Bases: Protocol

renew async 
renew(key: str, ttl_s: float) -> bool

True while this holder still owns key; False once ownership was lost.

LocalGuard 

LocalGuard()

Single-process exclusion. (Across a restart nothing is held, which is correct: the previous process is gone.)

LeaseGuard 

LeaseGuard(backend: object, member: str)

Cross-process exclusion via the bus backend's lease primitive.