Events¶

Events with a transactional outbox: an event is stored in the same transaction as the change that caused it, then published and delivered to its handlers in the background, at least once. One outbox, one relay and one set of background workers serve an application and the libraries it uses.

Quickstart¶

The outbox is one repository of the application's bundle and shares the unit of work with the others. Wire the background workers once and start them with the application:

from dataclasses import dataclass

from sqlalchemy.orm import sessionmaker
from sqlmodel import Session, SQLModel, create_engine

from commons.database import SqlUnitOfWorkFactory
from commons.events import CommitNotifyingUnitOfWorkFactory, Outbox, SqlOutbox, background_workers_from_settings


@dataclass
class Repositories:  # the app's bundle; anything with an `outbox` works
    invoices: InvoicesRepository
    outbox: Outbox


engine = create_engine("sqlite:///app.db")
SQLModel.metadata.create_all(engine)  # creates `outbox_events` with the app's tables

repositories = CommitNotifyingUnitOfWorkFactory(
    SqlUnitOfWorkFactory(
        sessionmaker(bind=engine, class_=Session),
        lambda session, user: Repositories(invoices=SqlInvoices(session), outbox=SqlOutbox(session)),
    )
)

workers = background_workers_from_settings(repositories)  # reads [app.events]
workers.subscribe(group="invoice-validation", event_type=InvoiceUploaded, handler=validate_invoice)

# In the application's lifespan, e.g. FastAPI:
#   workers.start()  on startup
#   await workers.stop()  on shutdown

CommitNotifyingUnitOfWorkFactory wakes the relay after every commit, so events are processed right away. Without it, the relay finds new events by polling.

Concepts¶

Event (Event): an immutable fact that happened, e.g. "an invoice was uploaded". A Pydantic model with a topic, an event_type and its own fields.
Envelope (EventEnvelope): the wire format of an event, routing metadata plus the JSON payload. Outbox and broker only see envelopes.
Topic: the log an event is appended to. Related events share a topic; a library uses topics of its own (e.g. conversations-runs).
Key: optional partition key; events with the same key are processed in order, events with different keys in parallel.
Outbox (Outbox, SqlOutbox): the table outbox_events the events are stored in, in the same transaction as the change that caused them. One per application.
Relay (OutboxRelay): publishes the stored events to the broker and marks them published.
Broker (LogBroker on an EventLog): stores the published events per topic and delivers them to the consumers. Kafka or Azure Service Bus could replace it.
Consumer group: a named reader of a topic with its own position (offset). Every group sees every event of the topics it subscribed to.
Handler (EventHandler): an async function that processes one typed event. It must be idempotent: it may see the same event twice.
Dead letter: an event whose handler kept failing; it is set aside so it cannot block its key.

Configuration¶

The background workers are configured under [app.events]. All settings are optional; unknown keys are rejected.

Setting	Type	Default	Description
`mode`	`"file"` \| `"memory"`	`"file"`	`file`: durable topic logs on disk; `memory`: lost on restart, for tests and development.
`dir`	path	`<data_dir>/events`	Directory of the `file` broker.
`poll_interval_seconds`	float	`5.0`	How often the relay and the consumers look for work they were not woken for.
`handler_max_concurrency`	int	`10`	Handlers of one consumer group running at the same time, for events with different keys.
`handler_max_attempts`	int	`3`	Attempts of a handler before the event is dead-lettered.
`outbox_retention_days`	int	`7`	How long published events are kept in the outbox.
`maintenance_interval_seconds`	float	`60.0`	How often the outbox is pruned and the maintenance jobs run.

[app.events]
mode = "file"
poll_interval_seconds = 5
handler_max_attempts = 3

Usage¶

Define an event¶

Derive from Event, set topic and event_type, and add the fields the handler needs. Keep events small: reference entities by id. Override key if the events of one entity must be processed in order.

from typing import ClassVar

from commons.events import Event


class InvoiceUploaded(Event):
    """An invoice was uploaded and waits to be validated."""

    topic: ClassVar[str] = "invoices"
    event_type: ClassVar[str] = "invoice_uploaded"

    invoice_id: str
    customer_id: str

    @property
    def key(self) -> str:
        """The invoices of one customer are validated in the order they were uploaded."""
        return self.customer_id

Name events in the past tense or as a request that happened (InvoiceUploaded, RunRequested), not as commands.
event_type must be unique within its topic and never change: it is stored in the outbox and the broker.
Add optional fields only; never rename or remove one. For an incompatible change, define a new event type.

Publish an event¶

Add the event to the outbox in the same unit of work as the change. Never publish to the broker directly.

async def upload_invoice(repositories, customer_id: str, data: bytes) -> str:
    with repositories.for_system() as (repos, commit):
        invoice = repos.invoices.add(customer_id, data)
        repos.outbox.add_event(InvoiceUploaded(invoice_id=invoice.id, customer_id=customer_id))
        await commit()  # stores invoice and event together; wakes the relay

    return invoice.id

If the unit of work is rolled back, neither the invoice nor the event is stored.

Handle an event¶

A handler is an async function that takes the typed event. Subscribe it with a consumer group before starting the workers. Raising makes the consumer retry the event; after handler_max_attempts it is dead-lettered.

async def validate_invoice(event: InvoiceUploaded) -> None:
    invoice = ...  # load the invoice by event.invoice_id
    if invoice is None or invoice.status != "uploaded":
        return  # redelivered or outdated event: nothing to do

    ...  # validate it, store the outcome


workers.subscribe(group="invoice-validation", event_type=InvoiceUploaded, handler=validate_invoice)

Be idempotent. Check the state before acting, or use event.event_id to detect duplicates.
Load current state. The event says what happened; the database says what is true now.
Raise only for transient errors that a retry can fix (e.g. a network error). Record business failures in the domain state and return normally.
One handler per event type and group. Use separate groups for independent reactions to the same event.
Persist long work step by step, each step in its own short unit of work, so no transaction stays open during slow calls; chain steps with further events.

Libraries¶

A library adds its events to the application's outbox: its repository bundle gets the outbox from the application's session (e.g. SqlOutbox(session)), and it offers a function that subscribes its handlers, e.g. subscribe_processes(workers, registry). The application calls it before starting the workers.

Add maintenance jobs¶

Periodic jobs, e.g. failing work left behind by a crash, run at start and every maintenance_interval_seconds, after the outbox is pruned:

workers = background_workers_from_settings(repositories, maintenance=[fail_stale_runs])

Test asynchronous code¶

Use EventsSettings(mode="memory") for the workers and SQLite in memory for the outbox.
Test handlers directly by calling them with an event; they are plain async functions.
Test that a domain function adds the event: read repos.outbox.list_unpublished(...) after it ran.
For end-to-end tests, start the workers, call the domain function and wait until the handler's effect is visible.

Behavior¶

At least once. Every committed event is delivered to every subscribed group at least once; after a crash it may be delivered again.
Order per key. Within a topic, a group processes the events of one key one after the other, in the order they were committed. Events with different keys, and events without a key, are processed in parallel, at most handler_max_concurrency at a time.
Redelivery after a crash. The committed offset only moves past events that are all processed: every unfinished event is delivered again, and so are finished events after an unfinished one.
A failing event only holds up its own key, until it is dead-lettered after handler_max_attempts. Unreadable events (the payload no longer validates) are dead-lettered right away.
Events nobody subscribed to are skipped; a group subscribed later does not see them.
Processing starts after the commit. Within the domain function the work is not done yet; return the pending state to the caller.
Crashed tasks restart. A background task that crashes is logged and restarted; stop never waits longer than its timeout.
Single application instance. The relay and LogBroker assume one running instance; several would publish events twice.

Errors¶

Exception	Raised when	What to do
`ValueError`	`subscribe` is called twice for the same group and event type.	Use one handler per event type and group, or a second group.
`RuntimeError`	`BackgroundWorkers.subscribe` is called after `start`, or `start` is called twice.	Subscribe all handlers before starting the workers.
`ValueError`	`Event.from_envelope` gets an envelope of another type, or one whose payload doesn't validate.	Only happens when converting envelopes by hand.
`ValueError`	`FileEventLog` gets a topic or group name that is not a safe file name.	Use letters, digits, `.`, `_` and `-`.
`ValidationError`	`[app.events]` has an unknown key or an invalid value.	Fix the settings.

API reference¶

commons.events ¶

Events with a transactional outbox: stored with the change that caused them, delivered at least once.

Events (Event, EventEnvelope): immutable facts with a topic, a type and an optional ordering key.
Ports (Outbox, HasOutbox, EventPublisher, EventSubscriber, EventConsumer, EventHandler).
Broker (LogBroker on an EventLog: InMemoryEventLog, FileEventLog): in-process, Kafka-like, single instance.
SQL outbox (SqlOutbox, SqlOutboxEvent): the table outbox_events, written in the caller's unit of work.
Relay and workers (OutboxRelay, BackgroundWorkers, CommitNotifyingUnitOfWorkFactory): publish committed events and run the handlers in the application's lifespan, configured by EventsSettings ([app.events]) and wired with background_workers_from_settings.

Modules:

Name	Description
`broker`
`event`
`log`
`ports`
`relay`
`settings`
`sql`
`workers`

Type Aliases:

Name	Description
`CommitListener`	Called after a successful commit, e.g. `OutboxRelay.wake`.
`EventHandler`	Processes one typed event; raising makes the consumer retry the event.
`Maintenance`	A periodic maintenance job, e.g. failing stale runs.

Classes:

Name	Description
`LogBroker`	An in-process message broker with Kafka-like semantics on top of an `EventLog`.
`Event`	Base of all events: an immutable fact, published via the outbox.
`EventEnvelope`	The wire format of an `Event`: routing metadata plus the JSON payload.
`DeadLetter`	An event a consumer group failed to process, with the error.
`EventLog`	The storage of a `LogBroker`: append-only topic logs plus the offsets of the consumer groups.
`FileEventLog`	An `EventLog` in plain files below `root`, durable across restarts.
`InMemoryEventLog`	An `EventLog` in process memory, for tests and development; everything is lost on restart.
`EventConsumer`	Consumes events from the message broker, e.g. Kafka consumer groups.
`EventPublisher`	Appends events to the message broker, e.g. a Kafka producer.
`EventSubscriber`	Registers the handlers of events, e.g. an `EventConsumer` or the background workers.
`HasOutbox`	A repository bundle of a unit of work that contains the outbox, e.g. an app's `Repositories` dataclass.
`Outbox`	The transactional outbox: events stored together with the change that caused them.
`OutboxRelay`	Publishes the events of the outbox and marks them published.
`EventsSettings`	The settings of the message broker, the outbox relay and the maintenance, read from `[app.events]`.
`SqlOutbox`	The SQL adapter of the `Outbox` port, on the table `outbox_events`.
`SqlOutboxEvent`	The persistence row of one outbox event (`outbox_events`).
`BackgroundWorkers`	The background tasks of the (single) application instance, started and stopped by its lifespan.
`CommitNotifyingUnitOfWorkFactory`	Wraps a `UnitOfWorkFactory` to call listeners after every successful commit.

Functions:

Name	Description
`background_workers_from_settings`	Wire the message broker, the outbox relay and the event consumer.

all `module-attribute` ¶

__all__ = [
    "BackgroundWorkers",
    "CommitListener",
    "CommitNotifyingUnitOfWorkFactory",
    "DeadLetter",
    "Event",
    "EventConsumer",
    "EventEnvelope",
    "EventHandler",
    "EventLog",
    "EventPublisher",
    "EventSubscriber",
    "EventsSettings",
    "FileEventLog",
    "HasOutbox",
    "InMemoryEventLog",
    "LogBroker",
    "Maintenance",
    "Outbox",
    "OutboxRelay",
    "SqlOutbox",
    "SqlOutboxEvent",
    "background_workers_from_settings",
]

EventHandler ¶

EventHandler = Callable[[E], Awaitable[object]]

Processes one typed event; raising makes the consumer retry the event.

CommitListener ¶

CommitListener = Callable[[], None]

Called after a successful commit, e.g. OutboxRelay.wake.

Maintenance ¶

Maintenance = Callable[[], Awaitable[object]]

A periodic maintenance job, e.g. failing stale runs.

LogBroker ¶

LogBroker(
    log: EventLog,
    *,
    poll_interval: float = 1.0,
    batch_size: int = 100,
    max_concurrency: int = 10,
    max_attempts: int = 3,
    retry_backoff: float = 1.0
)

An in-process message broker with Kafka-like semantics on top of an EventLog.

Implements EventPublisher and EventConsumer. Publishing appends to the topic's log and wakes the consumers of that topic. Each (group, topic) subscription reads its topic in order from the group's committed offset and processes the events in order per key: events with the same key one after the other, events with different keys (and events without a key) in parallel, at most max_concurrency handlers at a time and at most batch_size events in flight.

The committed offset only moves past events that are all processed, so after a crash every unfinished event is delivered again, and some finished ones too: at least once. A failing handler is tried max_attempts times in total, then the event is dead-lettered and skipped; an event that does not validate against its type is dead-lettered right away. Either way only its key waits, not the others. Consumers also poll every poll_interval seconds, in case an append bypassed publish.

Single process only: a Kafka or Service Bus adapter would replace this.

Parameters:

Name	Type	Description	Default
`log`	`EventLog`	The storage of the topics, offsets and dead letters.	required
`poll_interval`	`float`	Seconds between reads of a topic when the consumer was not woken by a publish.	`1.0`
`batch_size`	`int`	The maximum number of events read at once, and in flight (running or waiting for their key) per subscription.	`100`
`max_concurrency`	`int`	The maximum number of handlers running at the same time per subscription.	`10`
`max_attempts`	`int`	How often a handler is tried before the event is dead-lettered.	`3`
`retry_backoff`	`float`	Seconds to wait before a retry, multiplied by the number of the failed attempt.	`1.0`

Methods:

Name	Description
`publish`
`subscribe`
`run`

Source code in src/commons/events/broker.py

def __init__(
    self,
    log: EventLog,
    *,
    poll_interval: float = 1.0,
    batch_size: int = 100,
    max_concurrency: int = 10,
    max_attempts: int = 3,
    retry_backoff: float = 1.0,
) -> None:
    self._log = log
    self._poll_interval = poll_interval
    self._batch_size = batch_size
    self._max_concurrency = max_concurrency
    self._max_attempts = max_attempts
    self._retry_backoff = retry_backoff
    self._subscriptions: dict[tuple[str, str], _Subscription] = {}
    self._loop: asyncio.AbstractEventLoop | None = None
    self._wake: dict[tuple[str, str], asyncio.Event] = {}

publish `async` ¶

publish(envelope: EventEnvelope) -> None

Source code in src/commons/events/broker.py

async def publish(self, envelope: EventEnvelope) -> None:
    self._log.append(envelope)

    loop = self._loop
    for (_, topic), wake in list(self._wake.items()):
        if loop is not None and topic == envelope.topic:
            with suppress(RuntimeError):  # loop closed meanwhile
                loop.call_soon_threadsafe(wake.set)

subscribe ¶

subscribe(
    *,
    group: str,
    event_type: type[E],
    handler: EventHandler[E]
) -> None

Source code in src/commons/events/broker.py

def subscribe[E: Event](self, *, group: str, event_type: type[E], handler: EventHandler[E]) -> None:
    subscription = self._subscriptions.setdefault((group, event_type.topic), _Subscription(group, event_type.topic))
    if event_type.event_type in subscription.handlers:
        raise ValueError(f"Group `{group}` already handles `{event_type.topic}/{event_type.event_type}`.")

    subscription.handlers[event_type.event_type] = (event_type, handler)

run `async` ¶

run() -> None

Source code in src/commons/events/broker.py

async def run(self) -> None:
    self._loop = asyncio.get_running_loop()
    self._wake = {key: asyncio.Event() for key in self._subscriptions}
    try:
        await asyncio.gather(
            *(self._consume(subscription, self._wake[key]) for key, subscription in self._subscriptions.items())
        )
    finally:
        self._loop = None
        self._wake = {}

Event ¶

Bases: BaseModel

Base of all events: an immutable fact, published via the outbox.

Subclasses set topic (the log the event is appended to) and event_type (its name on the wire, unique within the topic), and may override key: events with the same key are processed in order, events with different keys in parallel. Delivery is at least once, so handlers must be idempotent; event_id is stable across redeliveries.

Attributes:

Name	Type	Description
`event_id`	`str`	The unique id of the event, a time-ordered UUIDv7 by default.
`occurred_at`	`datetime`	When the event occurred (UTC).

Example

class RunRequested(Event):
    topic: ClassVar[str] = "runs"
    event_type: ClassVar[str] = "run_requested"

    conversation_id: str
    run_id: str

    @property
    def key(self) -> str:
        return self.conversation_id

Methods:

Name	Description
`to_envelope`	Return the event in its wire format.
`from_envelope`	Return the typed event of an envelope.

model_config `class-attribute` `instance-attribute` ¶

model_config = ConfigDict(frozen=True)

topic `class-attribute` ¶

topic: str

event_type `class-attribute` ¶

event_type: str

event_id `class-attribute` `instance-attribute` ¶

event_id: str = Field(
    default_factory=new_id,
    min_length=ID_LENGTH,
    max_length=ID_LENGTH,
)

occurred_at `class-attribute` `instance-attribute` ¶

occurred_at: datetime = Field(default_factory=utc_now)

key `property` ¶

key: str | None

The partition key: events with the same key keep their order. None means no ordering guarantee.

to_envelope ¶

to_envelope() -> EventEnvelope

Return the event in its wire format.

Source code in src/commons/events/event.py

def to_envelope(self) -> EventEnvelope:
    """Return the event in its wire format."""
    return EventEnvelope(
        event_id=self.event_id,
        topic=self.topic,
        key=self.key,
        event_type=self.event_type,
        occurred_at=self.occurred_at,
        payload=self.model_dump(mode="json"),
    )

from_envelope `classmethod` ¶

from_envelope(envelope: EventEnvelope) -> Self

Return the typed event of an envelope.

Raises:

Type	Description
`ValueError`	If the envelope carries another event type, or its payload does not validate against this class (Pydantic's `ValidationError` is a `ValueError`).

Source code in src/commons/events/event.py

@classmethod
def from_envelope(cls, envelope: EventEnvelope) -> Self:
    """Return the typed event of an envelope.

    Raises:
        ValueError: If the envelope carries another event type, or its payload does not validate against this
            class (Pydantic's ``ValidationError`` is a ``ValueError``).
    """
    if (envelope.topic, envelope.event_type) != (cls.topic, cls.event_type):
        raise ValueError(
            f"Envelope `{envelope.event_id}` is `{envelope.topic}/{envelope.event_type}`, "
            f"not `{cls.topic}/{cls.event_type}`."
        )

    return cls.model_validate(envelope.payload)

EventEnvelope ¶

Bases: BaseModel

The wire format of an Event: routing metadata plus the JSON payload.

The outbox, the relay and the brokers only handle envelopes, so they need no knowledge of the concrete event classes. Consumers turn an envelope back into its typed event with Event.from_envelope.

Attributes:

Name	Type	Description
`event_id`	`str`	The unique id of the event; stable across redeliveries.
`topic`	`str`	The log the event is appended to.
`key`	`str \| None`	The partition key: events with the same key keep their order. None means no ordering guarantee.
`event_type`	`str`	The name of the event on the wire, unique within its topic.
`occurred_at`	`datetime`	When the event occurred (UTC).
`payload`	`dict[str, Any]`	The event's fields as JSON.

model_config `class-attribute` `instance-attribute` ¶

model_config = ConfigDict(frozen=True)

event_id `instance-attribute` ¶

event_id: str

topic `instance-attribute` ¶

topic: str

key `class-attribute` `instance-attribute` ¶

key: str | None = None

event_type `instance-attribute` ¶

event_type: str

occurred_at `instance-attribute` ¶

occurred_at: datetime

payload `instance-attribute` ¶

payload: dict[str, Any]

DeadLetter `dataclass` ¶

DeadLetter(group: str, envelope: EventEnvelope, error: str)

An event a consumer group failed to process, with the error.

Attributes:

Name	Type	Description
`group`	`str`
`envelope`	`EventEnvelope`
`error`	`str`

group `instance-attribute` ¶

group: str

envelope `instance-attribute` ¶

envelope: EventEnvelope

error `instance-attribute` ¶

error: str

EventLog ¶

Bases: Protocol

The storage of a LogBroker: append-only topic logs plus the offsets of the consumer groups.

An offset is an opaque position in a topic's log: 0 is its start, and every event read comes with the offset right after it, the one to commit once the event is processed.

Methods:

Name	Description
`append`	Durably append the envelope to the log of its topic.
`read`	Return at most `limit` events of `topic` from `offset` on, each with its next offset.
`committed_offset`	Return the offset `group` committed for `topic`; `0` if it never did.
`commit_offset`	Store `offset` as the position `group` continues `topic` from.
`dead_letter`	Keep an envelope that `group` failed to process, for manual inspection.

append ¶

append(envelope: EventEnvelope) -> None

Durably append the envelope to the log of its topic.

Source code in src/commons/events/log.py

def append(self, envelope: EventEnvelope) -> None:
    """Durably append the envelope to the log of its topic."""
    ...

read ¶

read(
    topic: str, offset: int, limit: int
) -> list[tuple[EventEnvelope, int]]

Return at most limit events of topic from offset on, each with its next offset.

Source code in src/commons/events/log.py

def read(self, topic: str, offset: int, limit: int) -> list[tuple[EventEnvelope, int]]:
    """Return at most ``limit`` events of ``topic`` from ``offset`` on, each with its next offset."""
    ...

committed_offset ¶

committed_offset(group: str, topic: str) -> int

Return the offset group committed for topic; 0 if it never did.

Source code in src/commons/events/log.py

def committed_offset(self, group: str, topic: str) -> int:
    """Return the offset ``group`` committed for ``topic``; ``0`` if it never did."""
    ...

commit_offset ¶

commit_offset(group: str, topic: str, offset: int) -> None

Store offset as the position group continues topic from.

Source code in src/commons/events/log.py

def commit_offset(self, group: str, topic: str, offset: int) -> None:
    """Store ``offset`` as the position ``group`` continues ``topic`` from."""
    ...

dead_letter ¶

dead_letter(
    group: str, envelope: EventEnvelope, error: str
) -> None

Keep an envelope that group failed to process, for manual inspection.

Source code in src/commons/events/log.py

def dead_letter(self, group: str, envelope: EventEnvelope, error: str) -> None:
    """Keep an envelope that ``group`` failed to process, for manual inspection."""
    ...

FileEventLog ¶

FileEventLog(root: Path)

An EventLog in plain files below root, durable across restarts.

<root>/<topic>.jsonl                          one envelope per line, append-only
<root>/_offsets/<group>/<topic>.json          committed offset of the group
<root>/_dead_letters/<group>/<topic>.jsonl    events the group failed to process

The offset of an event is the byte position after its line. A line without its trailing newline (an append cut short by a crash) is never read.

Single process only: appends and offset files are not locked.

Parameters:

Name	Type	Description	Default
`root`	`Path`	The directory of the log.	required

Methods:

Name	Description
`append`
`read`
`committed_offset`
`commit_offset`
`dead_letter`

Attributes:

Name	Type	Description
`root`

Source code in src/commons/events/log.py

def __init__(self, root: Path) -> None:
    self.root = root

root `instance-attribute` ¶

root = root

append ¶

append(envelope: EventEnvelope) -> None

Source code in src/commons/events/log.py

def append(self, envelope: EventEnvelope) -> None:
    path = self._topic_path(envelope.topic)
    path.parent.mkdir(parents=True, exist_ok=True)
    line = envelope.model_dump_json() + "\n"

    with path.open("ab") as file:
        file.write(line.encode("utf-8"))
        file.flush()
        os.fsync(file.fileno())

read ¶

read(
    topic: str, offset: int, limit: int
) -> list[tuple[EventEnvelope, int]]

Source code in src/commons/events/log.py

def read(self, topic: str, offset: int, limit: int) -> list[tuple[EventEnvelope, int]]:
    path = self._topic_path(topic)
    if not path.exists():
        return []

    events: list[tuple[EventEnvelope, int]] = []
    with path.open("rb") as file:
        file.seek(offset)
        while len(events) < limit:
            line = file.readline()
            if not line.endswith(b"\n"):
                break
            offset += len(line)
            events.append((EventEnvelope.model_validate_json(line), offset))

    return events

committed_offset ¶

committed_offset(group: str, topic: str) -> int

Source code in src/commons/events/log.py

def committed_offset(self, group: str, topic: str) -> int:
    path = self._offset_path(group, topic)
    if not path.exists():
        return 0

    return int(json.loads(path.read_text())["offset"])

commit_offset ¶

commit_offset(group: str, topic: str, offset: int) -> None

Source code in src/commons/events/log.py

def commit_offset(self, group: str, topic: str, offset: int) -> None:
    path = self._offset_path(group, topic)
    path.parent.mkdir(parents=True, exist_ok=True)
    temp = path.with_suffix(".tmp")
    temp.write_text(json.dumps({"offset": offset}))
    temp.replace(path)  # atomic: a crash leaves the old or the new offset

dead_letter ¶

dead_letter(
    group: str, envelope: EventEnvelope, error: str
) -> None

Source code in src/commons/events/log.py

def dead_letter(self, group: str, envelope: EventEnvelope, error: str) -> None:
    path = self.root / "_dead_letters" / _checked(group) / f"{_checked(envelope.topic)}.jsonl"
    path.parent.mkdir(parents=True, exist_ok=True)
    entry = {
        "dead_lettered_at": utc_now().isoformat(),
        "error": error,
        "envelope": envelope.model_dump(mode="json"),
    }

    with path.open("a", encoding="utf-8") as file:
        file.write(json.dumps(entry) + "\n")

InMemoryEventLog ¶

InMemoryEventLog()

An EventLog in process memory, for tests and development; everything is lost on restart.

The offset of an event is its index in the topic's list, plus one.

Methods:

Name	Description
`append`
`read`
`committed_offset`
`commit_offset`
`dead_letter`

Attributes:

Name	Type	Description
`topics`	`dict[str, list[EventEnvelope]]`
`offsets`	`dict[tuple[str, str], int]`
`dead_letters`	`list[DeadLetter]`

Source code in src/commons/events/log.py

def __init__(self) -> None:
    self.topics: dict[str, list[EventEnvelope]] = {}
    self.offsets: dict[tuple[str, str], int] = {}
    self.dead_letters: list[DeadLetter] = []

topics `instance-attribute` ¶

topics: dict[str, list[EventEnvelope]] = {}

offsets `instance-attribute` ¶

offsets: dict[tuple[str, str], int] = {}

dead_letters `instance-attribute` ¶

dead_letters: list[DeadLetter] = []

append ¶

append(envelope: EventEnvelope) -> None

Source code in src/commons/events/log.py

def append(self, envelope: EventEnvelope) -> None:
    self.topics.setdefault(envelope.topic, []).append(envelope)

read ¶

read(
    topic: str, offset: int, limit: int
) -> list[tuple[EventEnvelope, int]]

Source code in src/commons/events/log.py

def read(self, topic: str, offset: int, limit: int) -> list[tuple[EventEnvelope, int]]:
    events = self.topics.get(topic, [])[offset : offset + limit]
    return [(envelope, offset + i + 1) for i, envelope in enumerate(events)]

committed_offset ¶

committed_offset(group: str, topic: str) -> int

Source code in src/commons/events/log.py

def committed_offset(self, group: str, topic: str) -> int:
    return self.offsets.get((group, topic), 0)

commit_offset ¶

commit_offset(group: str, topic: str, offset: int) -> None

Source code in src/commons/events/log.py

def commit_offset(self, group: str, topic: str, offset: int) -> None:
    self.offsets[(group, topic)] = offset

dead_letter ¶

dead_letter(
    group: str, envelope: EventEnvelope, error: str
) -> None

Source code in src/commons/events/log.py

def dead_letter(self, group: str, envelope: EventEnvelope, error: str) -> None:
    self.dead_letters.append(DeadLetter(group, envelope, error))

EventConsumer ¶

Bases: EventSubscriber, Protocol

Consumes events from the message broker, e.g. Kafka consumer groups.

Each consumer group reads every topic it subscribed to from its own committed offset and processes the events in order per key (Event.key): events with the same key one after the other, events with different keys, or without a key, possibly in parallel. Handlers must not rely on an order across keys. An offset is committed only once the event and all events before it were processed: delivery is at least once. An event whose handler keeps failing is dead-lettered, so it cannot block its key. Events of types the group has no handler for are skipped.

Methods:

Name	Description
`run`	Consume all subscriptions until cancelled.

run `async` ¶

run() -> None

Consume all subscriptions until cancelled.

Source code in src/commons/events/ports.py

async def run(self) -> None:
    """Consume all subscriptions until cancelled."""
    ...

EventPublisher ¶

Bases: Protocol

Appends events to the message broker, e.g. a Kafka producer.

Only the outbox relay publishes; domain functions add events to the Outbox instead.

Methods:

Name	Description
`publish`	Durably append the envelope to its topic; raise if it was not accepted.

publish `async` ¶

publish(envelope: EventEnvelope) -> None

Durably append the envelope to its topic; raise if it was not accepted.

Source code in src/commons/events/ports.py

async def publish(self, envelope: EventEnvelope) -> None:
    """Durably append the envelope to its topic; raise if it was not accepted."""
    ...

EventSubscriber ¶

Bases: Protocol

Registers the handlers of events, e.g. an EventConsumer or the background workers.

Methods:

Name	Description
`subscribe`	Let `group` process the events of `event_type` with `handler`.

subscribe ¶

subscribe(
    *,
    group: str,
    event_type: type[E],
    handler: EventHandler[E]
) -> None

Let group process the events of event_type with handler.

Raises:

Type	Description
`ValueError`	If `group` already has a handler for `event_type`.

Source code in src/commons/events/ports.py

def subscribe[E: Event](self, *, group: str, event_type: type[E], handler: EventHandler[E]) -> None:
    """Let ``group`` process the events of ``event_type`` with ``handler``.

    Raises:
        ValueError: If ``group`` already has a handler for ``event_type``.
    """
    ...

HasOutbox ¶

Bases: Protocol

A repository bundle of a unit of work that contains the outbox, e.g. an app's Repositories dataclass.

The relay needs nothing else of the bundle, so any app's or library's bundle with an outbox field works.

Attributes:

Name	Type	Description
`outbox`	`Outbox`	The outbox, sharing the unit of work's transaction.

outbox `property` ¶

outbox: Outbox

The outbox, sharing the unit of work's transaction.

Outbox ¶

Bases: Protocol

The transactional outbox: events stored together with the change that caused them.

An outbox shares the unit of work of the other repositories, so an added event is stored if and only if the change is committed. The outbox relay later publishes the stored events and marks them published.

Methods:

Name	Description
`add_event`	Store the event for publishing once the unit of work commits.
`list_unpublished`	Return at most `limit` unpublished events, in the order they were added.
`mark_published`	Mark the events with the given ids published; unknown ids are ignored.
`delete_published_before`	Delete the events published before `before`; return how many were deleted.

add_event ¶

add_event(event: Event) -> None

Store the event for publishing once the unit of work commits.

Source code in src/commons/events/ports.py

def add_event(self, event: Event) -> None:
    """Store the event for publishing once the unit of work commits."""
    ...

list_unpublished ¶

list_unpublished(limit: int) -> list[EventEnvelope]

Return at most limit unpublished events, in the order they were added.

Source code in src/commons/events/ports.py

def list_unpublished(self, limit: int) -> list[EventEnvelope]:
    """Return at most ``limit`` unpublished events, in the order they were added."""
    ...

mark_published ¶

mark_published(
    event_ids: list[str], published_at: datetime
) -> None

Mark the events with the given ids published; unknown ids are ignored.

Source code in src/commons/events/ports.py

def mark_published(self, event_ids: list[str], published_at: datetime) -> None:
    """Mark the events with the given ids published; unknown ids are ignored."""
    ...

delete_published_before ¶

delete_published_before(before: datetime) -> int

Delete the events published before before; return how many were deleted.

Source code in src/commons/events/ports.py

def delete_published_before(self, before: datetime) -> int:
    """Delete the events published before ``before``; return how many were deleted."""
    ...

OutboxRelay ¶

OutboxRelay(
    repositories: UnitOfWorkFactory[HasOutbox],
    publisher: EventPublisher,
    *,
    poll_interval: float = 5.0,
    batch_size: int = 100,
    retention: timedelta = timedelta(days=7)
)

Publishes the events of the outbox and marks them published.

The only component that publishes: run relays everything unpublished at start (events left over from a crash), then again whenever wake is called (after a commit) or poll_interval seconds passed. An event is marked published only after the broker accepted it, so a crash in between publishes it again: at least once. Events are published in the order they were added.

Single application instance only: several relays would publish the same events. With several instances, claim rows with SELECT ... FOR UPDATE SKIP LOCKED (or elect one leader) before publishing.

Parameters:

Name	Type	Description	Default
`repositories`	`UnitOfWorkFactory[HasOutbox]`	Opens the units of work to read and update the outbox with.	required
`publisher`	`EventPublisher`	The broker the events are published to.	required
`poll_interval`	`float`	Seconds between relays when the relay was not woken.	`5.0`
`batch_size`	`int`	The maximum number of events read from the outbox at once.	`100`
`retention`	`timedelta`	How long published events are kept in the outbox before `prune` deletes them.	`timedelta(days=7)`

Methods:

Name	Description
`wake`	Relay now instead of at the next poll; safe to call from any thread.
`run`	Relay until cancelled; a failed relay is logged and retried at the next poll.
`relay_pending`	Publish all unpublished events, oldest first; return how many were published.
`prune`	Delete the events published longer ago than the retention; return how many.

Source code in src/commons/events/relay.py

def __init__(
    self,
    repositories: UnitOfWorkFactory[HasOutbox],
    publisher: EventPublisher,
    *,
    poll_interval: float = 5.0,
    batch_size: int = 100,
    retention: timedelta = timedelta(days=7),
) -> None:
    self._repositories = repositories
    self._publisher = publisher
    self._poll_interval = poll_interval
    self._batch_size = batch_size
    self._retention = retention
    self._loop: asyncio.AbstractEventLoop | None = None
    self._wake: asyncio.Event | None = None

wake ¶

wake() -> None

Relay now instead of at the next poll; safe to call from any thread.

Source code in src/commons/events/relay.py

def wake(self) -> None:
    """Relay now instead of at the next poll; safe to call from any thread."""

    loop, wake = self._loop, self._wake
    if loop is not None and wake is not None:
        with suppress(RuntimeError):  # loop closed meanwhile
            loop.call_soon_threadsafe(wake.set)

run `async` ¶

run() -> None

Relay until cancelled; a failed relay is logged and retried at the next poll.

Source code in src/commons/events/relay.py

async def run(self) -> None:
    """Relay until cancelled; a failed relay is logged and retried at the next poll."""

    self._loop = asyncio.get_running_loop()
    self._wake = wake = asyncio.Event()
    try:
        while True:
            wake.clear()  # before relaying, so a commit during it is not missed
            try:
                await self.relay_pending()
            except Exception:
                LOG.exception("Relaying outbox events failed; retrying")

            with suppress(TimeoutError):
                await asyncio.wait_for(wake.wait(), self._poll_interval)
    finally:
        self._loop = self._wake = None

relay_pending `async` ¶

relay_pending() -> int

Publish all unpublished events, oldest first; return how many were published.

Source code in src/commons/events/relay.py

async def relay_pending(self) -> int:
    """Publish all unpublished events, oldest first; return how many were published."""

    total = 0

    while True:
        with self._repositories.for_system() as (repositories, _):
            batch = repositories.outbox.list_unpublished(self._batch_size)

        if not batch:
            return total

        published: list[str] = []
        try:
            for envelope in batch:
                await self._publisher.publish(envelope)
                published.append(envelope.event_id)
        finally:
            if published:
                with self._repositories.for_system() as (repositories, commit):
                    repositories.outbox.mark_published(published, utc_now())
                    await commit()
                total += len(published)

prune `async` ¶

prune() -> int

Delete the events published longer ago than the retention; return how many.

Source code in src/commons/events/relay.py

async def prune(self) -> int:
    """Delete the events published longer ago than the retention; return how many."""

    with self._repositories.for_system() as (repositories, commit):
        deleted = repositories.outbox.delete_published_before(utc_now() - self._retention)
        await commit()

    return deleted

EventsSettings ¶

Bases: BaseModel

The settings of the message broker, the outbox relay and the maintenance, read from [app.events].

Attributes:

Name	Type	Description
`mode`	`Literal['file', 'memory']`	`file`: durable topic logs below `dir`; `memory`: lost on restart (tests, development).
`dir`	`Path \| None`	The directory of the `file` broker; defaults to `<data_dir>/events`.
`poll_interval_seconds`	`float`	How often the relay and the consumers look for work they were not woken for.
`handler_max_concurrency`	`int`	How many event handlers of one consumer group may run at the same time, for events with different keys.
`handler_max_attempts`	`int`	How often an event handler is tried before the event is dead-lettered.
`outbox_retention_days`	`int`	How long published events are kept in the outbox.
`maintenance_interval_seconds`	`float`	How often the outbox is pruned and the maintenance jobs run.

Example

[app.events]
mode = "file"
poll_interval_seconds = 5
handler_max_attempts = 3

Methods:

Name	Description
`events_dir`	Return the directory of the `file` broker.
`read`	Read the settings from `[app.events]`; the defaults if the section is missing.

model_config `class-attribute` `instance-attribute` ¶

model_config = ConfigDict(extra='forbid')

mode `class-attribute` `instance-attribute` ¶

mode: Literal['file', 'memory'] = 'file'

dir `class-attribute` `instance-attribute` ¶

dir: Path | None = None

poll_interval_seconds `class-attribute` `instance-attribute` ¶

poll_interval_seconds: float = Field(default=5.0, gt=0)

handler_max_concurrency `class-attribute` `instance-attribute` ¶

handler_max_concurrency: int = Field(default=10, gt=0)

handler_max_attempts `class-attribute` `instance-attribute` ¶

handler_max_attempts: int = Field(default=3, gt=0)

outbox_retention_days `class-attribute` `instance-attribute` ¶

outbox_retention_days: int = Field(default=7, gt=0)

maintenance_interval_seconds `class-attribute` `instance-attribute` ¶

maintenance_interval_seconds: float = Field(
    default=60.0, gt=0
)

outbox_retention `property` ¶

outbox_retention: timedelta

events_dir ¶

events_dir() -> Path

Return the directory of the file broker.

Source code in src/commons/events/settings.py

def events_dir(self) -> Path:
    """Return the directory of the ``file`` broker."""

    return self.dir or AppSettings.read().data_dir / "events"

read `staticmethod` ¶

read() -> EventsSettings

Read the settings from [app.events]; the defaults if the section is missing.

Source code in src/commons/events/settings.py

@staticmethod
def read() -> "EventsSettings":
    """Read the settings from ``[app.events]``; the defaults if the section is missing."""

    try:
        return read_settings_object("events", EventsSettings)
    except KeyError:
        return EventsSettings()

SqlOutbox ¶

SqlOutbox(session: Session)

The SQL adapter of the Outbox port, on the table outbox_events.

The outbox never commits; the caller (the unit of work) owns the transaction, so an added event is stored together with the change that caused it.

Parameters:

Name	Type	Description	Default
`session`	`Session`	The session of the unit of work.	required

Methods:

Name	Description
`add_event`
`list_unpublished`
`mark_published`
`delete_published_before`

Attributes:

Name	Type	Description
`session`

Source code in src/commons/events/sql.py

def __init__(self, session: Session) -> None:
    self.session = session

session `instance-attribute` ¶

session = session

add_event ¶

add_event(event: Event) -> None

Source code in src/commons/events/sql.py

def add_event(self, event: Event) -> None:
    self.session.add(SqlOutboxEvent.from_envelope(event.to_envelope()))

list_unpublished ¶

list_unpublished(limit: int) -> list[EventEnvelope]

Source code in src/commons/events/sql.py

def list_unpublished(self, limit: int) -> list[EventEnvelope]:
    stmt = (
        select(SqlOutboxEvent)
        .where(col(SqlOutboxEvent.published_at).is_(None))
        .order_by(col(SqlOutboxEvent.sequence))
        .limit(limit)
    )
    return [row.to_envelope() for row in self.session.exec(stmt).all()]

mark_published ¶

mark_published(
    event_ids: list[str], published_at: datetime
) -> None

Source code in src/commons/events/sql.py

def mark_published(self, event_ids: list[str], published_at: datetime) -> None:
    if not event_ids:
        return

    self.session.execute(
        update(SqlOutboxEvent).where(col(SqlOutboxEvent.event_id).in_(event_ids)).values(published_at=published_at)
    )

delete_published_before ¶

delete_published_before(before: datetime) -> int

Source code in src/commons/events/sql.py

def delete_published_before(self, before: datetime) -> int:
    result: CursorResult = self.session.execute(  # type: ignore[assignment]
        delete(SqlOutboxEvent).where(col(SqlOutboxEvent.published_at) < before)
    )
    return result.rowcount

SqlOutboxEvent ¶

Bases: SQLModel

The persistence row of one outbox event (outbox_events).

Not historized: the outbox is plumbing, not domain state. sequence (an autoincrement key) keeps the order the events were added in; published_at is NULL until the relay published the event. One table per application: the app and the libraries it uses add their events to it, on their own topics.

Methods:

Name	Description
`from_envelope`	Return the row of an envelope.
`to_envelope`	Return the envelope of the row.

Attributes:

Name	Type	Description
`__tablename__`
`sequence`	`int \| None`
`event_id`	`str`
`topic`	`str`
`key`	`str \| None`
`event_type`	`str`
`occurred_at`	`datetime`
`payload`	`dict[str, Any]`
`published_at`	`datetime \| None`

tablename `class-attribute` `instance-attribute` ¶

__tablename__ = 'outbox_events'

sequence `class-attribute` `instance-attribute` ¶

sequence: int | None = Field(default=None, primary_key=True)

event_id `class-attribute` `instance-attribute` ¶

event_id: str = Field(unique=True, max_length=ID_LENGTH)

topic `class-attribute` `instance-attribute` ¶

topic: str = Field(max_length=CODE_LENGTH)

key `class-attribute` `instance-attribute` ¶

key: str | None = Field(
    default=None, max_length=NAME_LENGTH
)

event_type `class-attribute` `instance-attribute` ¶

event_type: str = Field(max_length=CODE_LENGTH)

occurred_at `class-attribute` `instance-attribute` ¶

occurred_at: datetime = Field(sa_type=UTCDateTime)

payload `class-attribute` `instance-attribute` ¶

payload: dict[str, Any] = Field(sa_type=JSON)

published_at `class-attribute` `instance-attribute` ¶

published_at: datetime | None = Field(
    default=None, sa_type=UTCDateTime, index=True
)

from_envelope `staticmethod` ¶

from_envelope(envelope: EventEnvelope) -> SqlOutboxEvent

Return the row of an envelope.

Source code in src/commons/events/sql.py

@staticmethod
def from_envelope(envelope: EventEnvelope) -> "SqlOutboxEvent":
    """Return the row of an envelope."""
    return SqlOutboxEvent(
        event_id=envelope.event_id,
        topic=envelope.topic,
        key=envelope.key,
        event_type=envelope.event_type,
        occurred_at=envelope.occurred_at,
        payload=envelope.payload,
    )

to_envelope ¶

to_envelope() -> EventEnvelope

Return the envelope of the row.

Source code in src/commons/events/sql.py

def to_envelope(self) -> EventEnvelope:
    """Return the envelope of the row."""
    return EventEnvelope(
        event_id=self.event_id,
        topic=self.topic,
        key=self.key,
        event_type=self.event_type,
        occurred_at=_utc(self.occurred_at),
        payload=self.payload,
    )

BackgroundWorkers ¶

BackgroundWorkers(
    relay: OutboxRelay,
    consumer: EventConsumer,
    *,
    maintenance: list[Maintenance] | None = None,
    maintenance_interval: float = 60.0
)

The background tasks of the (single) application instance, started and stopped by its lifespan.

the outbox relay, publishing committed events,
the event consumer, executing the handlers subscribed with subscribe,
maintenance: pruning the outbox and the given maintenance jobs, at start and every maintenance_interval seconds.

A task that crashes is logged and restarted, so one failure cannot silently stop event processing.

Parameters:

Name	Type	Description	Default
`relay`	`OutboxRelay`	Publishes the events of the outbox.	required
`consumer`	`EventConsumer`	Delivers the published events to their handlers.	required
`maintenance`	`list[Maintenance] \| None`	Further periodic jobs, run after the outbox is pruned.	`None`
`maintenance_interval`	`float`	Seconds between maintenance runs.	`60.0`

Methods:

Name	Description
`subscribe`	Let consumer `group` process the events of `event_type` with `handler`.
`start`	Start all tasks on the running event loop.
`stop`	Cancel all tasks and wait until they ended, at most `timeout` seconds.
`maintain`	Prune the outbox and run the maintenance jobs once.

Source code in src/commons/events/workers.py

def __init__(
    self,
    relay: OutboxRelay,
    consumer: EventConsumer,
    *,
    maintenance: list[Maintenance] | None = None,
    maintenance_interval: float = 60.0,
) -> None:
    self._relay = relay
    self._consumer = consumer
    self._maintenance = list(maintenance or [])
    self._maintenance_interval = maintenance_interval
    self._tasks: list[asyncio.Task] = []

subscribe ¶

subscribe(
    *,
    group: str,
    event_type: type[E],
    handler: EventHandler[E]
) -> None

Let consumer group process the events of event_type with handler.

Raises:

Type	Description
`RuntimeError`	If the workers are already running; subscribe before `start`.
`ValueError`	If `group` already has a handler for `event_type`.

Source code in src/commons/events/workers.py

def subscribe[E: Event](self, *, group: str, event_type: type[E], handler: EventHandler[E]) -> None:
    """Let consumer ``group`` process the events of ``event_type`` with ``handler``.

    Raises:
        RuntimeError: If the workers are already running; subscribe before `start`.
        ValueError: If ``group`` already has a handler for ``event_type``.
    """

    if self._tasks:
        raise RuntimeError("Subscribe before the background workers are started.")

    self._consumer.subscribe(group=group, event_type=event_type, handler=handler)

start ¶

start() -> None

Start all tasks on the running event loop.

Source code in src/commons/events/workers.py

def start(self) -> None:
    """Start all tasks on the running event loop."""

    if self._tasks:
        raise RuntimeError("Background workers are already running.")

    self._tasks = [
        asyncio.create_task(_supervised(name, work), name=name)
        for name, work in [
            ("outbox-relay", self._relay.run),
            ("event-consumer", self._consumer.run),
            ("maintenance", self._maintain),
        ]
    ]

stop `async` ¶

stop(timeout: float = _STOP_TIMEOUT_SECONDS) -> None

Cancel all tasks and wait until they ended, at most timeout seconds.

A task still running is cancelled again every half second: a library in a handler (e.g. anyio in an LLM client) may swallow a cancellation, and the task's loop then carries on. Tasks left after timeout are logged and abandoned, so a stuck task never blocks the shutdown.

Source code in src/commons/events/workers.py

async def stop(self, timeout: float = _STOP_TIMEOUT_SECONDS) -> None:
    """Cancel all tasks and wait until they ended, at most ``timeout`` seconds.

    A task still running is cancelled again every half second: a library in a handler (e.g. anyio in an LLM
    client) may swallow a cancellation, and the task's loop then carries on. Tasks left after ``timeout`` are
    logged and abandoned, so a stuck task never blocks the shutdown.
    """

    tasks, self._tasks = self._tasks, []
    pending = set(tasks)
    deadline = asyncio.get_running_loop().time() + timeout

    while pending:
        for task in pending:
            task.cancel()
        _, pending = await asyncio.wait(pending, timeout=0.5)

        if pending and asyncio.get_running_loop().time() >= deadline:
            names = ", ".join(sorted(task.get_name() for task in pending))
            LOG.error(f"Background tasks did not stop within {timeout}s: {names}")
            return

maintain `async` ¶

maintain() -> None

Prune the outbox and run the maintenance jobs once.

Source code in src/commons/events/workers.py

async def maintain(self) -> None:
    """Prune the outbox and run the maintenance jobs once."""

    await self._relay.prune()
    for job in self._maintenance:
        await job()

CommitNotifyingUnitOfWorkFactory ¶

CommitNotifyingUnitOfWorkFactory(
    inner: UnitOfWorkFactory[T],
)

Bases: UnitOfWorkFactory[T]

Wraps a UnitOfWorkFactory to call listeners after every successful commit.

Lets the outbox relay publish events right after the transaction that stored them, without domain functions having to know about the relay. A failing listener is logged and never fails the (already committed) unit of work.

Parameters:

Name	Type	Description	Default
`inner`	`UnitOfWorkFactory[T]`	The wrapped factory.	required

Methods:

Name	Description
`add_commit_listener`	Call `listener` after every successful commit.
`for_user`
`for_system`
`__enter__`	Open a unit of work like entering the wrapped factory (for its current user), with notifying commit.
`__exit__`

Source code in src/commons/events/workers.py

def __init__(self, inner: UnitOfWorkFactory[T]) -> None:
    self._inner = inner
    self._listeners: list[CommitListener] = []

add_commit_listener ¶

add_commit_listener(listener: CommitListener) -> None

Call listener after every successful commit.

Source code in src/commons/events/workers.py

def add_commit_listener(self, listener: CommitListener) -> None:
    """Call ``listener`` after every successful commit."""

    self._listeners.append(listener)

for_user ¶

for_user(user: User) -> UnitOfWork[T]

Source code in src/commons/events/workers.py

def for_user(self, user: User) -> UnitOfWork[T]:
    return _CommitNotifyingUnitOfWork(self._inner.for_user(user), self._listeners)

for_system ¶

for_system() -> UnitOfWork[T]

Source code in src/commons/events/workers.py

def for_system(self) -> UnitOfWork[T]:
    return _CommitNotifyingUnitOfWork(self._inner.for_system(), self._listeners)

enter ¶

__enter__() -> tuple[T, CommitFunction]

Open a unit of work like entering the wrapped factory (for its current user), with notifying commit.

Source code in src/commons/events/workers.py

def __enter__(self) -> tuple[T, CommitFunction]:
    """Open a unit of work like entering the wrapped factory (for its current user), with notifying commit."""
    repositories, commit = self._inner.__enter__()
    return repositories, _notifying(commit, self._listeners)

exit ¶

__exit__(*args: object) -> None

Source code in src/commons/events/workers.py

def __exit__(self, *args: object) -> None:
    self._inner.__exit__(*args)

background_workers_from_settings ¶

background_workers_from_settings(
    repositories: UnitOfWorkFactory[HasOutbox],
    settings: EventsSettings | None = None,
    maintenance: list[Maintenance] | None = None,
) -> BackgroundWorkers

Wire the message broker, the outbox relay and the event consumer.

The broker is selected by [app.events] mode. If repositories signals commits (a CommitNotifyingUnitOfWorkFactory), the relay publishes right after every commit; otherwise it relies on polling. Subscribe the event handlers with BackgroundWorkers.subscribe before starting the workers.

Parameters:

Name	Type	Description	Default
`repositories`	`UnitOfWorkFactory[HasOutbox]`	Opens the units of work whose repositories contain the outbox.	required
`settings`	`EventsSettings \| None`	The settings. Read from `[app.events]` if not given.	`None`
`maintenance`	`list[Maintenance] \| None`	Further periodic jobs, run after the outbox is pruned.	`None`

Source code in src/commons/events/settings.py

def background_workers_from_settings(
    repositories: UnitOfWorkFactory[HasOutbox],
    settings: EventsSettings | None = None,
    maintenance: list[Maintenance] | None = None,
) -> BackgroundWorkers:
    """Wire the message broker, the outbox relay and the event consumer.

    The broker is selected by ``[app.events] mode``. If ``repositories`` signals commits (a
    `CommitNotifyingUnitOfWorkFactory`), the relay publishes right after every commit; otherwise it relies on
    polling. Subscribe the event handlers with `BackgroundWorkers.subscribe` before starting the workers.

    Args:
        repositories: Opens the units of work whose repositories contain the outbox.
        settings: The settings. Read from ``[app.events]`` if not given.
        maintenance: Further periodic jobs, run after the outbox is pruned.
    """

    settings = settings or EventsSettings.read()

    log: EventLog = FileEventLog(settings.events_dir()) if settings.mode == "file" else InMemoryEventLog()
    broker = LogBroker(
        log,
        poll_interval=settings.poll_interval_seconds,
        max_concurrency=settings.handler_max_concurrency,
        max_attempts=settings.handler_max_attempts,
    )
    relay = OutboxRelay(
        repositories,
        broker,
        poll_interval=settings.poll_interval_seconds,
        retention=settings.outbox_retention,
    )

    if isinstance(repositories, CommitNotifyingUnitOfWorkFactory):
        repositories.add_commit_listener(relay.wake)

    return BackgroundWorkers(
        relay,
        broker,
        maintenance=maintenance,
        maintenance_interval=settings.maintenance_interval_seconds,
    )

Events¶

Quickstart¶

Concepts¶

Configuration¶

Usage¶

Define an event¶

Publish an event¶

Handle an event¶

Libraries¶

Add maintenance jobs¶

Test asynchronous code¶

Behavior¶

Errors¶

API reference¶

commons.events ¶

__all__ module-attribute ¶

EventHandler ¶

CommitListener ¶

Maintenance ¶

LogBroker ¶

publish async ¶

subscribe ¶

run async ¶

Event ¶

model_config class-attribute instance-attribute ¶

topic class-attribute ¶

event_type class-attribute ¶

event_id class-attribute instance-attribute ¶

occurred_at class-attribute instance-attribute ¶

key property ¶

to_envelope ¶

from_envelope classmethod ¶

EventEnvelope ¶

model_config class-attribute instance-attribute ¶

event_id instance-attribute ¶

topic instance-attribute ¶

key class-attribute instance-attribute ¶

event_type instance-attribute ¶

occurred_at instance-attribute ¶

payload instance-attribute ¶

DeadLetter dataclass ¶

group instance-attribute ¶

envelope instance-attribute ¶

error instance-attribute ¶

EventLog ¶

append ¶

read ¶

committed_offset ¶

commit_offset ¶

dead_letter ¶

FileEventLog ¶

root instance-attribute ¶

append ¶

read ¶

committed_offset ¶

commit_offset ¶

dead_letter ¶

InMemoryEventLog ¶

topics instance-attribute ¶

offsets instance-attribute ¶

dead_letters instance-attribute ¶

append ¶

read ¶

committed_offset ¶

commit_offset ¶

dead_letter ¶

EventConsumer ¶

run async ¶

EventPublisher ¶

publish async ¶

EventSubscriber ¶

subscribe ¶

HasOutbox ¶

outbox property ¶

Outbox ¶

add_event ¶

list_unpublished ¶

mark_published ¶

delete_published_before ¶

OutboxRelay ¶

all `module-attribute` ¶

publish `async` ¶

run `async` ¶

model_config `class-attribute` `instance-attribute` ¶

topic `class-attribute` ¶

event_type `class-attribute` ¶

event_id `class-attribute` `instance-attribute` ¶

occurred_at `class-attribute` `instance-attribute` ¶

key `property` ¶

from_envelope `classmethod` ¶

model_config `class-attribute` `instance-attribute` ¶

event_id `instance-attribute` ¶

topic `instance-attribute` ¶

key `class-attribute` `instance-attribute` ¶

event_type `instance-attribute` ¶

occurred_at `instance-attribute` ¶

payload `instance-attribute` ¶

DeadLetter `dataclass` ¶

group `instance-attribute` ¶

envelope `instance-attribute` ¶

error `instance-attribute` ¶

root `instance-attribute` ¶

topics `instance-attribute` ¶

offsets `instance-attribute` ¶

dead_letters `instance-attribute` ¶

run `async` ¶

publish `async` ¶

outbox `property` ¶

run `async` ¶

relay_pending `async` ¶

prune `async` ¶

model_config `class-attribute` `instance-attribute` ¶

mode `class-attribute` `instance-attribute` ¶

dir `class-attribute` `instance-attribute` ¶

poll_interval_seconds `class-attribute` `instance-attribute` ¶

handler_max_concurrency `class-attribute` `instance-attribute` ¶

handler_max_attempts `class-attribute` `instance-attribute` ¶

outbox_retention_days `class-attribute` `instance-attribute` ¶

maintenance_interval_seconds `class-attribute` `instance-attribute` ¶

outbox_retention `property` ¶

read `staticmethod` ¶

session `instance-attribute` ¶

tablename `class-attribute` `instance-attribute` ¶

sequence `class-attribute` `instance-attribute` ¶

event_id `class-attribute` `instance-attribute` ¶

topic `class-attribute` `instance-attribute` ¶

key `class-attribute` `instance-attribute` ¶

event_type `class-attribute` `instance-attribute` ¶

occurred_at `class-attribute` `instance-attribute` ¶

payload `class-attribute` `instance-attribute` ¶

published_at `class-attribute` `instance-attribute` ¶

from_envelope `staticmethod` ¶

stop `async` ¶

maintain `async` ¶

enter ¶

exit ¶