API reference¶
Everything below is exported from django_domain_events directly.
Declaring¶
event ¶
event(
cls: type | None = None, *, name: str | None = None, version: int = 1
) -> type | Callable[[type], type]
Register a frozen dataclass as an event, bare or called.
The default name is <app_label>.<ClassName>. Pin it with name= when
renaming the class would otherwise strand rows written under the old one.
receiver ¶
receiver(
event_class: type[E],
*,
mode: DeliveryMode = DeliveryMode.DURABLE,
takes_context: bool = False,
key: str | None = None,
max_attempts: int = 5,
eager: bool = False,
site: str = "relay",
lease_seconds: int | None = None,
) -> Callable[[Callable[..., None]], Callable[..., None]]
Register a callable to receive one event type.
takes_context is the spelling django.tasks.task uses for the same
idea. The overloads make a checker enforce the arity it implies, so
declaring one and writing the other fails at the decorator rather than in
the relay hours later.
max_attempts is copied onto each delivery row at fire time.
site is the execution knob, separate from mode on purpose: timing is
what a receiver promises about the transaction, and where its code runs is a
different question that only a queue answers. "relay" runs it in the
relay worker; "task" hands it to the configured task backend, which then
acknowledges the row when it finishes.
eager additionally attempts delivery immediately after commit, in the
firing process, with the relay as the fallback for whatever process death
loses. It is what stops DURABLE feeling slow: outbox durability at
on-commit latency, at the cost of a duplicate when the process dies
mid-receiver - which at-least-once already required everyone to tolerate.
lease_seconds overrides LEASE_SECONDS for this receiver alone, and
is the answer for one that legitimately runs long. A receiver still working
when its lease lapses has its row taken by another worker and its own work
rolled back - correct, and entirely wasted. Declaring it here rather than
offering the receiver a way to extend its lease from the inside, because
that cannot work: the receiver runs inside the transaction that carries its
acknowledgement, so anything it writes is invisible to every other worker
until it has already finished.
Firing¶
fire ¶
fire(
event: object, *, dedupe_key: str | None = None, occurred_at: datetime | None = None
) -> int | None
Record an event and owe it to its durable receivers. Returns its id.
Returns the event id, or None when suppression discarded it without
recording.
This does not call durable receivers; it records intent. The event row and
one delivery row per durable receiver are written inside the caller's
transaction, so the obligation exists if and only if the business change
committed. A durable receiver therefore cannot signal failure back here -
only INLINE receivers can, by raising.
attributed ¶
attributed(
*,
actor: Any = None,
actor_key: str = "",
actor_label: str = "",
correlation_id: UUID | None = None,
**data: Any,
) -> Iterator[Scope]
Attach ambient facts to every event fired inside this block.
Nested blocks layer: an inner one overrides what it sets and inherits the rest, so a request-level actor survives a block that only adds a source. The actor is one such thing - naming any part of it replaces all of it, because a key from one block beside a user id from another describes two actors.
actor is any object; its identity is derived once, here. Only an instance
of the user model reaches the actor column, which is a foreign key to it;
everything else is identified by actor_key alone.
current_scope ¶
The scope in effect right now. Read at fire time, never later.
A fresh empty Scope when none is set, not a shared one: a singleton's dict would be handed to callers and to receivers, and one mutation would reach every later event in the process.
suppressed ¶
Fire these events without delivering them, and say why.
The reason is required and lands on the row. A silently dropped event is unauditable, which is the failure mode suppression is most likely to cause, so the default writes the event and marks it rather than discarding it.
record=False discards instead. It exists because a hundred-thousand-row
import writing a hundred thousand suppressed rows is a surprise, and it
trades the audit trail for the write - which is the whole point of the
default, so it is named rather than defaulted.
Nested blocks accumulate rather than replace. A library that suppresses its own event type inside your block must not re-enable yours: the innermost matching reason is the one recorded, and any matching block asking not to record wins, because that is the safer half of the disagreement.
caused_by ¶
Mark events fired inside this block as descended from event_id.
Set around every receiver, at every execution site, so an event a receiver fires records its parent with no ceremony at the call site - a parameter threaded through every receiver is a parameter someone forgets.
Both values come off the parent's row, never from a ContextVar: by the
time a durable delivery runs, the block that attributed the parent has long
exited, possibly in another process. Causation is one hop; the correlation
id is the whole tree, and carrying it here is what keeps a grandchild fired
hours later in the same chain as the request that started it.
propagate_scope ¶
Carry the current scope into a thread you start yourself.
threading.Thread and ThreadPoolExecutor.submit start with an empty
context, not a copy of yours - so a worker thread silently loses the
attribution of whoever spawned it. This is the gotcha that actually bites,
because nothing fails: events simply arrive with no actor.
executor.submit(propagate_scope(fn), *args)
Call it at submit time, not as a @propagate_scope decorator. It
captures the scope when it is called, and at decoration time - import time -
there is none, so the decorator form silently carries nothing, which is the
exact failure it exists to prevent.
It captures the scope's values rather than a contextvars.Context, so
the wrapper is reusable: one Context cannot be entered twice, and a fan-out
that submits the same wrapped callable per item would raise on the second.
Not needed across sync_to_async / async_to_sync, which carry context
both ways, nor for asyncio tasks, which inherit a copy at creation. And
it cannot help across a process boundary, where the answer is the event row.
Delivery¶
run_relay ¶
run_relay(
*,
worker_id: str,
passes: int | None = None,
now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
sleep: Callable[[float], None] = time_module.sleep,
wait: Callable[[float], bool] | None = None,
allow_unsafe_concurrency: bool = False,
) -> dict[DeliveryStatus, int]
Claim and deliver until passes is spent, or forever if it is None.
The clock and the sleep are arguments rather than module calls so the loop is testable without waiting: every branch here turns on time, and a suite that had to elapse real seconds to reach one would either be slow or never reach it.
Refuses to start where the database cannot express a skipped lock. Two
workers there would hand the same row to two receivers on every pass, which
at-least-once tolerates but nobody wants as a steady state.
allow_unsafe_concurrency lifts that for a deployment running exactly one
relay, which is a real shape in development; running two under it is the
thing the guard exists to prevent.
deliver_one ¶
Run one delivery and record its outcome. None means it was lost.
The receiver's work and the acknowledgement commit together, so a receiver touching only this database is effectively once: the duplicate at-least-once owes you cannot be observed. Side effects outside the database are at-least-once, as promised.
Every write here is a compare-and-set against the claim this call read, and
the lease is extended to cover this one delivery before the receiver runs.
Both exist for the same reason: a claim can lapse while its worker is still
alive, and a worker that has lost its row must not go on to write a verdict
over whoever legitimately took it. Losing the row returns None rather
than raising - it is an ordinary outcome of a lease expiring, not a fault.
deliver_pending ¶
deliver_pending(
limit: int | None = None,
*,
worker_id: str = "deliver_pending",
ignore_backoff: bool = False,
) -> dict[DeliveryStatus, int]
Claim and deliver what is owed, and report the outcome.
limit=None means everything owed, claimed in batches until none is left;
a number caps it at that many. Claims go through the same leased path the
relay uses, so a pass here and a running relay do not hand the same row to
two receivers - on a backend with row locking. SQLite has none, so two
concurrent passes there can both take the same row.
claim_batch ¶
claim_batch(
*,
worker_id: str,
now: datetime,
lease: timedelta,
limit: int,
only_ids: list[int] | None = None,
ignore_backoff: bool = False,
) -> list[int]
Take ownership of up to limit deliveries and return their ids.
Claims are leased, not marked: a worker that dies without acknowledging becomes re-claimable when its lease lapses, which is the same path as an ordinary retry rather than a special case.
Rows are selected by available_at and never by primary key. A
transaction holding a lower id can commit after one holding a higher id, so
a row can become visible "in the past"; a high-water mark would skip it
forever.
ignore_backoff claims rows whose retry is still scheduled. It exists for
the test helper, which cannot wait out a jittered hour to observe a retry.
backoff ¶
How long to wait before attempt number attempt may be retried.
jitter is supplied by the caller as a value in [0, 1) rather than drawn
here, so the schedule is a pure function of its arguments and a test can
assert the curve instead of sampling it.
Full jitter: the delay is drawn from the whole window up to the exponential ceiling, not added on top of it. Retrying a shared downstream at ceiling-plus-a-bit keeps every failed delivery in the same cohort, which is the thundering herd the backoff was meant to break up.
notify_relay ¶
Tell a listening relay that something is owed.
Fire-and-forget: a notification sent while nobody is listening is simply lost, which is why the relay's poll stays as the floor rather than being replaced by this. It removes latency and never carries the obligation - the delivery row does.
Operations¶
prune_events ¶
prune_events(
older_than: timedelta | None = None,
*,
now: datetime | None = None,
batch_size: int | None = None,
limit: int | None = None,
) -> int
Delete settled events older than the window, and return how many went.
An outbox without a prune story becomes the largest table in the database, and it becomes it quietly - which is why this ships rather than waiting for someone to notice.
Only settled events: one with a delivery still pending, failed or claimed is still owed, and deleting it would drop work nobody recorded as lost. An event with no delivery rows at all - suppressed, or fired with no durable receivers - is settled by definition.
Deletes in batches. A single statement over a year of rows takes a lock for as long as it runs, on the table the relay is trying to claim from.
replay_events ¶
replay_events(
event_ids: Iterable[int], *, receiver_keys: Iterable[str] | None = None
) -> dict[str, int]
Make these events owed again, and report what changed.
The receiver set freezes at fire time, so deploying a new receiver does not hand it a backlog of week-old events. That is deliberate - and this is the other half of it: replay is an operation somebody invokes, with a name, and not an accident of a deploy.
Two things happen, and they are counted separately because they are different decisions. A terminal delivery is reopened: it ran, and you want it to run again. A receiver with no row for the event is added: it did not exist when the event fired, and you are choosing to give it the backlog.
A delivery still in flight is left alone. Reopening a claimed row would hand the same work to two receivers, which is the one thing the lease exists to prevent.
requeue_dead ¶
requeue_dead(
*,
receiver_key: str | None = None,
delivery_ids: Iterable[int] | None = None,
limit: int | None = None,
) -> int
Give dead-lettered deliveries their attempt budget back.
Dead is where a delivery stops on its own; it is not where it stops for good. Attempts reset to zero rather than staying spent, because a row requeued at its limit dead-letters again on the first failure and the operator learns nothing they did not already know.
Scoped by receiver, because the usual reason to requeue is that one downstream was broken and now is not. Scoped by row as well, because the other reason is an operator reading a dead-letter list and picking the four they understand.
Introspection¶
catalogue ¶
Describe every declared event, its payload shape and its receivers.
The artefact a team wants and signals cannot produce: what exists, what listens, and under which guarantee - generated from the declarations rather than maintained beside them, so it cannot drift.
Sorted by name throughout. A catalogue is written to a file and diffed against the last one, and import order is not a difference.
render_catalogue ¶
Render a catalogue as a document, in Markdown or JSON.
Both, because they answer different questions: Markdown is read by a person onboarding onto a codebase, JSON is diffed by a pipeline that wants to fail a pull request for removing a field other teams consume.
Both end in exactly one newline. These are written to a file and diffed, and a file with no final newline reports a change on its last line forever.
what_listens_to ¶
Every receiver declared for one event, sorted by key.
The question signals cannot answer: a Django signal's receivers are a list of weak references keyed by an opaque dispatch uid, so "who reacts to this" is answerable only by grepping.
listens_for ¶
The event one receiver is declared for, or None if no such receiver.
The inverse direction, and the one an operator needs: a dead-letter row names a receiver key, and the next question is always what it was supposed to be receiving.
None covers two cases that look the same from a delivery row - a key never declared, and one whose event class was deleted out from under it.
quiet_receivers ¶
quiet_receivers(
*, within: timedelta | None = None, now: datetime | None = None
) -> list[QuietReceiver]
Declared receivers that have succeeded at nothing inside the window.
The query an event log makes possible and a signal never will: "this receiver has not run since June" is a fact here, not a guess, because every durable delivery left a row.
Driven by the registry rather than by the table, so a receiver that has never received anything appears - which is the answer worth having, and exactly the one a query over delivery rows alone cannot produce.
Only DURABLE receivers write rows, so only they are reported. An INLINE or ON_COMMIT receiver has no delivery history to be quiet about, and listing it as silent forever would train the reader to ignore the output.
Read off succeeded_at rather than completed_at, and with no status
filter at all. Both of those describe the current cycle: replay and
requeue reopen a row and clear them, so an operator who replays yesterday's
events would then be told the receiver had never run. Max ignores nulls,
so a receiver whose every delivery failed still reads as never having
succeeded without a predicate saying so.
The window defaults to RETENTION_DAYS, which is not a coincidence of numbers: past that point the prune has deleted the evidence, so "quiet for longer than retention" is the longest answer this can honestly give.
outbox_health ¶
How far behind the outbox is.
The gap the package left until now: quiet_receivers() answers whether a
receiver is running, and nothing answered whether the queue is draining.
Those fail differently - a relay that has been down for an hour has every
receiver quiet and a backlog climbing, while a single wedged receiver has a
backlog and everything else fine.
Owed means "not terminal", which is what the prune settles by and a superset of what the relay can claim right now - a row inside its backoff window is owed and not yet claimable. The superset is the useful side: this cannot report an empty queue while work is still outstanding.
Testing¶
drain_outbox ¶
drain_outbox(
limit: int | None = None, *, respect_backoff: bool = False
) -> dict[DeliveryStatus, int]
Deliver everything owed, from a test, through the production code path.
Deliberately not a task_always_eager equivalent: bypassing the transport
hides both the serialisation boundary and the timing, which is how a suite
passes while production breaks. This runs the same claim, encode, decode and
acknowledgement as the relay, and skips only the waiting.
Skipping the waiting includes the retry backoff, which is why this ignores
it by default: a failed delivery is scheduled a jittered interval ahead - up
to an hour - and a test cannot sit that out. Pass respect_backoff=True
to assert the schedule itself.
assert_fired ¶
Assert an event was fired, and return the decoded events.
Reads the log rather than patching fire: a mock records that a function
was called, while the row is what the rest of the system reacts to. Decoding
on the way out means a payload that cannot round-trip fails here too.
Types¶
DeliveryMode ¶
Bases: Enum
Timing and guarantee, declared per receiver.
Timing is separate from where a receiver's code runs; a queue is only ever an answer to the second question.
INLINE
class-attribute
instance-attribute
¶
Inside the firing transaction, and free to abort it by raising.
Needs no durability: its failure mode is a rollback, so nothing can be owed.
ON_COMMIT
class-attribute
instance-attribute
¶
After commit, in the firing process, best effort. Not recoverable.
DURABLE
class-attribute
instance-attribute
¶
After commit, at-least-once, retried, with a row recording the debt.
DeliveryStatus ¶
Bases: TextChoices
Status of one (event, durable receiver) pair.
DeliveryContext
dataclass
¶
Delivery metadata, passed only to receivers declaring takes_context.
Frozen data read off the event row, never a live handle: a durable delivery can run in another process hours after the scope that produced it has gone.
Scope
dataclass
¶
The ambient facts captured onto every event fired inside a block.
Frozen, and nested blocks build a new one rather than mutating: a scope that could be edited from inside would let a receiver rewrite the attribution of events fired after it.
merged ¶
Layer another scope over this one, keeping what it does not set.
Catalogue
dataclass
¶
Every declared event and what listens to it, at one moment.
A snapshot rather than a live view: it is built to be written to a file and compared against the one from last release, which is the whole point of having it.
CatalogueEvent
dataclass
¶
One declared event, its payload shape and everything listening to it.
migrates_older_rows
class-attribute
instance-attribute
¶
Whether the class declares upgrade. Worth publishing: it is the
difference between an event whose old rows still decode and one whose old
rows dead-letter after the next breaking change.
Last, and defaulted, so adding it does not break a consumer constructing one - these are exported types.
CatalogueField
dataclass
¶
One field of an event payload, as the catalogue describes it.
required
instance-attribute
¶
False when the field has a default, which is also the condition under which adding it is a non-breaking change to a log with rows already in it.
CatalogueReceiver
dataclass
¶
One receiver of one event, as the catalogue describes it.
lease_seconds
class-attribute
instance-attribute
¶
Defaulted so adding it does not break a consumer constructing one.
OutboxHealth
dataclass
¶
Whether the outbox is keeping up, in one value.
Counts and one timestamp rather than rates: a rate needs two samples and a place to keep the first, which is a metrics system's job. This is what that system scrapes.
oldest_owed_at
instance-attribute
¶
When the oldest still-owed delivery's event was recorded, or None when nothing is owed.
Not when it next becomes due: a failed delivery has its available_at
pushed into the future by the backoff, so an alert written against that
reads a negative age exactly while a receiver is failing. This rises
monotonically for as long as work sits undone, which is what a threshold
needs.
A receiver that fails all the way to dead leaves the owed set entirely, so
watch dead alongside it. One number does not cover both.
lapsed_leases
instance-attribute
¶
Claimed rows whose lease has expired. Steady non-zero means workers are dying mid-delivery, or a receiver outruns its lease and has its work thrown away every time - see lease_seconds= on the receiver.
receivers
instance-attribute
¶
Only receivers with something owed or dead, worst backlog first.
QuietReceiver
dataclass
¶
A declared receiver with nothing delivered to it inside the window.
last_succeeded_at
instance-attribute
¶
None means never - which is the more interesting answer of the two.
ReceiverBacklog
dataclass
¶
What one receiver is behind on.
oldest_owed_at
instance-attribute
¶
When the oldest still-owed delivery's event was recorded. See
OutboxHealth.oldest_owed_at for why this is not when it becomes due.
RegisteredEvent
dataclass
¶
RegisteredReceiver
dataclass
¶
lease_seconds
class-attribute
instance-attribute
¶
None means the LEASE_SECONDS setting. Defaulted because it is the one field of a declaration that is genuinely optional: every other value here is something the decorator always resolves.
TaskBackend ¶
Bases: Protocol
Where a durable receiver's code runs when the relay hands it off.
The relay claims the row either way; this only decides who executes it. That is the whole reason the execution site is a separate knob from the timing: a queue is only ever an answer to the second question.
The backend may lose an enqueue without consequence. The delivery row is the record, so anything dropped is reclaimed when the lease lapses - which is what makes a lossy queue safe here, and what keeps this a small protocol.
PayloadCodec ¶
Bases: Protocol
How an event instance becomes a payload, and comes back.
A seam because the two halves cost differently: encoding needs nothing this package does not have, while decoding a nested payload back into a frozen dataclass is worth a dependency. Every codec consumes an ordinary frozen dataclass, so the choice never reaches how events are declared.
PayloadUpgradeFailed ¶
Bases: Exception
An event class's upgrade hook raised while migrating an old row.
Its own type because an operator reading last_error needs to know the
hook ran and what it said, not that something unspecified went wrong
somewhere between the row and the receiver.
At the package root rather than under codecs/: it is raised before the
codec sees the payload, and importing it from there pulls in
codecs/__init__, which imports the module that imports this one.