API reference

Everything below is exported from django_domain_events directly.

Declaring

event

event(cls: E) -> E

event(
    *,
    name: str | None = None,
    version: int = 1,
    retention: timedelta | Retention | None = None,
) -> Callable[[E], E]

event(
    cls: type | None = None,
    *,
    name: str | None = None,
    version: int = 1,
    retention: timedelta | Retention | None = None,
) -> 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. A name may not be longer than the column that holds it, 255 characters, whether given or derived: a longer one is refused here with ValueError, because Postgres raises at fire() time while SQLite enforces nothing.

retention says how long prune_events keeps this event's rows: a timedelta for a window of its own, Retention.SUCCEEDED or Retention.SETTLED to delete it once consumed, or None for RETENTION_DAYS. fire() copies it onto every row it records, so a later change here does not reach events already recorded.

receiver

receiver(
    event_class: type[E],
    *,
    mode: DeliveryMode = DeliveryMode.DURABLE,
    takes_context: Literal[False] = False,
    key: str | None = None,
    max_attempts: int = 5,
    eager: bool = False,
    site: str = "relay",
    lease_seconds: int | None = None,
    on_failure: Callable[[DeliveryFailure], Awaitable[None] | None] | None = None,
    targets: Targets[E] | None = None,
    backoff_base_seconds: float | None = None,
    backoff_cap_seconds: float | None = None,
    lane: str = DEFAULT_LANE,
    give_up_after: timedelta | None = None,
    due: timedelta | Due[E] | None = None,
) -> Callable[[Plain[E]], Plain[E]]

receiver(
    event_class: type[E],
    *,
    mode: DeliveryMode = DeliveryMode.DURABLE,
    takes_context: Literal[True],
    key: str | None = None,
    max_attempts: int = 5,
    eager: bool = False,
    site: str = "relay",
    lease_seconds: int | None = None,
    on_failure: Callable[[DeliveryFailure], Awaitable[None] | None] | None = None,
    targets: Targets[E] | None = None,
    backoff_base_seconds: float | None = None,
    backoff_cap_seconds: float | None = None,
    lane: str = DEFAULT_LANE,
    give_up_after: timedelta | None = None,
    due: timedelta | Due[E] | None = None,
) -> Callable[[WithContext[E]], WithContext[E]]

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,
    on_failure: Callable[[DeliveryFailure], Awaitable[None] | None] | None = None,
    targets: Targets[E] | None = None,
    backoff_base_seconds: float | None = None,
    backoff_cap_seconds: float | None = None,
    lane: str = DEFAULT_LANE,
    give_up_after: timedelta | None = None,
    due: timedelta | Due[E] | None = None,
) -> Callable[[Callable[..., None]], Callable[..., None]]

Register a callable to receive one event type, or every event.

event_class may be AnyEvent, which declares a wildcard: the receiver is owed every event fired, including events declared by apps that load after it. See AnyEvent for what that is for and what it costs.

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.

key is written onto every delivery row of a DURABLE receiver and is what the row is found by, so it may not be longer than the column that holds it, 255 characters. A longer one - explicit, or derived from a very long function name - is refused here with ValueError, because the database either truncates it, orphaning the delivery, or raises at fire time; SQLite enforces nothing, so it would pass a test suite and fail in production. The rule holds at every mode, though INLINE and ON_COMMIT store no key, so that a key stays valid when its mode changes.

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.

The receiver may be declared async def, at every mode. It is run to completion where a function would be called - in fire() for INLINE, at commit for ON_COMMIT, in the relay, a task or an eager attempt for DURABLE - through asgiref.sync.async_to_sync, while the calling thread serves its sync_to_async calls, so Django's async ORM writes land in the transaction carrying the acknowledgement and roll back with it. Outside ASGI each call runs on a fresh event loop of its own, closed when it returns; under ASGI an INLINE or ON_COMMIT coroutine runs on the server's loop. Either way a client bound to a loop cannot be shared between calls. What it buys is the ability to call async libraries, not concurrency: the relay still runs one delivery at a time. A coroutine that is cancelled is a failed attempt. A sync callable that returns an awaitable - a functools.wraps decorator around an async def - or an async def containing yield is refused at the call with TypeError, rather than recorded as done with its body never run; mark such a wrapper with asgiref.sync.markcoroutinefunction. on_failure may be async def on the same terms. targets and due may not; see below.

on_failure is called after a failed attempt has been recorded, outside the transaction that was just rolled back, so a receiver can keep a durable record of its own failure. It could not before: everything a receiver writes commits with its acknowledgement and is discarded the moment it raises, so the attempts worth logging were exactly the ones that could not be logged. The hook must not raise; one that does is logged and swallowed, because a failure path that fails is worse than a lost log line.

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. Where every durable receiver in its lane declares the same lease, it is also the lease the lane's claims stamp - at most LEASE_SECONDS, plus 10 percent of headroom - so it changes how big a batch the lane can work through before the batch's tail lapses.

targets makes the receiver a fan-out: a callable taking the event and a DeliveryContext, returning the strings this event is owed to - endpoint ids, tenant slugs, anything that names a destination. fire() writes one delivery row per target instead of one per receiver, each with its own attempt count, backoff and dead-letter, and hands the target to the receiver as DeliveryContext.target. A target returned twice is delivered once, and a callable returning nothing writes no row at all, which is how a wildcard receiver says "not this event". Each target must be a non-empty string; anything else is refused where it is returned.

The callable runs inside the caller's transaction, at fire time, exactly as the event row is written - so fire()'s transactional contract is its contract too. What it reads is what the business change can see, including that change's own uncommitted rows. If it raises, fire() raises, and the caller's transaction fails with it: the change, the event and every delivery row roll back together. That is deliberate. A fan-out that swallowed the error and delivered to nobody would commit a change whose consequences silently never happened, which is the failure this package exists to rule out. Keep the callable's failure modes in view: it is code in the middle of somebody else's write.

It is also a query in the hot path. fire() calls every fan-out receiver's callable on every event it is owed - for a wildcard, every event fired - so a callable that queries a table costs one query per fired event per fan-out receiver. For a transport replacing an event per destination that is a clear saving; for one declared carelessly it is a cost paid by every write in the system. Make the lookup indexed, and make it return early for events it can rule out without one.

replay_events calls it again, so a replay goes to the targets that exist at replay time.

It must be a plain function, not async def, and so must a due callable: each would cost an event loop per receiver per fire, inside the caller's request, for what is a database read. Either is refused here.

backoff_base_seconds and backoff_cap_seconds override BACKOFF_BASE_SECONDS and BACKOFF_CAP_SECONDS for this receiver alone, as lease_seconds overrides LEASE_SECONDS. They are read when an attempt fails, not copied onto the row, so a curve changed by a deploy applies to deliveries already in flight. Either may be declared without the other, which then comes from its setting; only when both are declared is the cap refused for being below the base, since a setting can change after the declaration is read. Full jitter applies to this curve as to the default: each wait is drawn from zero up to the ceiling, so a base of sixty seconds means a first retry somewhere in the next minute, possibly after a few seconds, never "in a minute". A list of delays or a callable is not offered, because the catalogue could not publish one.

lane names the relay processes that claim this receiver's rows: deliver_events --lane mail claims only receivers declared with lane="mail", and a relay started without --lane claims every row no named lane takes - including rows whose receiver has since been deleted, which it records orphaned. It is read from the registry at claim time, so moving a receiver to another lane moves its owed rows with it. It is how a slow receiver stops holding up the rest, and it is declared here rather than as relay flags because the default relay is the one that must exclude it, and an exclusion list kept in deployment manifests goes stale. An eager=True attempt still runs in the firing process, whatever the lane.

give_up_after is the bound that ends a delivery whose receiver defers with RetryAfter(seconds, counts=False). That deferral spends no attempt, so max_attempts cannot end it; this does. Once a row has been owed for this long - since it was written, or since a replay or requeue reopened it - the next deferral dead-letters it, as a spent budget would. It is a timedelta because a bare number has no unit. Without it, a deferral that does not count is counted after all, with a warning, so that every delivery still ends. It bounds deferrals alone: an ordinary failure or a counting RetryAfter is still ended by max_attempts, however long the row has been owed.

due makes the receiver a scheduled reaction: its delivery of an event is written now and runs later. A timedelta counts from the event's occurred_at, the domain's own time; a callable is handed the event and returns an aware datetime, for a time the event itself names. The time belongs to the receiver rather than to fire() because one event drives reactions at different times - send the offer now, expire it in 48 hours.

The callable runs inside the caller's transaction, at fire time, as a targets callable does, and if it raises, or returns anything but an aware datetime, fire() raises: a reaction that silently never happened is the failure this package exists to rule out. A naive datetime is refused rather than read as UTC or as local time, either of which is wrong for somebody by hours. A due time already past - an event fired with an old occurred_at - is due as soon as the row is written, and owed from then.

The row is written with its available_at and its due_at at the due time, so the relay claims it then and not before; DeliveryContext.due_at hands the time to the receiver, so it can recompute the due time from current state and answer an early run with NotYet. Everything else about the delivery is unchanged - attempts, backoff, dead letters, site="task", lanes - and starts at the due time. replay_events resolves it again and writes the later of then and now.

Refused beside a non-durable mode, which has no row to wait in; beside eager=True, which means "attempt it at commit"; and beside targets, because a fan-out resolved at fire time would keep that recipient set until the delivery came due while a replay resolves it when it runs. A scheduled receiver that needs a fan-out fires a follow-up event whose targets resolve when it is due.

AnyEvent

Declare a receiver for every event: @receiver(AnyEvent, ...).

A marker, never fired and never instantiated. registry.receivers_for returns the receivers declared for an event's class plus every receiver declared for this one, so a wildcard is matched when an event is fired - not when the receiver is declared. That is the whole difference from walking the registry at startup and declaring a receiver per event, which silently misses every event declared by an app that loads afterwards.

Meant for a transport - something that forwards events without knowing what they are, such as a webhook sender or a bridge to a broker. A wildcard on its own writes a delivery row for every event the system fires, which is why it is usually declared with targets=: a callable returning nothing for an event nobody wants writes no row at all.

The receiver is handed whatever was fired, so its event parameter is honestly typed object.

It appears in the catalogue in a section of its own rather than under every event, and what_listens_to(OrderPlaced) leaves it out; ask what_listens_to(AnyEvent) for the wildcards.

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.

A durable receiver declared with targets= gets one row per target instead, and its callable runs here, inside the caller's transaction. If it raises, this raises, and the caller's change rolls back with the event: a fan-out that delivered to nobody would be a committed change whose consequences silently never happened. It also costs whatever the callable costs, on every event it is owed - one query per fired event per fan-out receiver, for one that reads a table.

A durable receiver declared with due= has its row written due later: available_at and due_at at its due time, never earlier than now. A due callable runs here too, on the same terms as targets: if it raises, or answers with anything but an aware datetime, this raises.

The event's declared retention is copied onto the row. A dedupe_key is refused, with ValueError, for an event declared to delete on consumption: the key is unique on the row, so it would protect only until the prune deleted it.

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

current_scope() -> 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

suppressed(*event_classes: type, reason: str, record: bool = True) -> Iterator[None]

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

caused_by(event_id: int, correlation_id: UUID | None = None) -> Iterator[None]

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

propagate_scope(func: Callable[..., R]) -> Callable[..., R]

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] = timezone.now,
    sleep: Callable[[float], None] = time_module.sleep,
    wait: Callable[[float], bool] | None = None,
    stop: Callable[[], bool] = lambda: False,
    allow_unsafe_concurrency: bool = False,
    lane: str | None = DEFAULT_LANE,
    batch_size: int | None = None,
    monotonic: Callable[[], float] = time_module.monotonic,
) -> dict[DeliveryStatus, int]

Claim and deliver until passes is spent, stop says so, or forever.

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.

stop is read before every claim and before every delivery. When it returns true the relay finishes the delivery in hand, hands back the rest of its claimed batch, and returns: the rows it gives back are claimable by any worker at once, rather than when their lease lapses. An idle relay reads it when its wait ends, so it stops within POLL_SECONDS. deliver_events wires it to SIGTERM and SIGINT.

The relay survives its database. A claim, a delivery or a wait that fails is logged and the loop goes on; after a database error the connection is closed so the next query opens a fresh one, and a failed claim is retried after a backoff that doubles from POLL_SECONDS up to 15 seconds.

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.

lane is the lane this process serves: only the rows of receivers declared with that lane=, read from the registry on every claim. The default lane is every row no named lane takes, including the rows of a receiver that has since been deleted, and it is what a relay serves unless told otherwise - the default relay staying out of a slow lane is the whole of the isolation. None serves every lane, for a single relay in development. A lane no receiver is declared in is refused at start, because a relay serving it would claim nothing and look healthy doing it.

Concurrency is the number of processes. Deliveries run one at a time within a relay, so a lane that must go faster runs more relays with the same lane; nothing here limits the rate across them.

batch_size sizes this relay's claims in place of BATCH_SIZE, which also sizes prune batches and requeue chunks. Every row of a batch is claimed under one lease, and a row still waiting its turn when that lease lapses is taken by another relay, so a slow lane wants a batch it can work through inside the lease its claim stamps. That is the lane's own lease plus 10 percent where every durable receiver in it shares one no longer than LEASE_SECONDS, and LEASE_SECONDS plus 10 percent otherwise: 66 seconds, not LEASE_SECONDS, for a lane of receivers declaring lease_seconds=60. A row that starts while its claimed lease is still its receiver's own starts without extending it, which saves a write per delivery, so a batch whose rows all start inside a tenth of the lease costs the fewest.

A relay that finds nothing to claim also prunes, so an event declared to be deleted on consumption goes within minutes without a prune_events cron line. The sweep is prune_events() with its defaults, run from the idle branch at most once per RELAY_PRUNE_SECONDS, the first one an interval after the relay starts rather than at its first idle pass: a relay that is restarted often, or run for a few passes, then never sweeps at all. Every relay sweeps whatever its lane. Nothing is lost or deleted twice, because the prune re-checks at the delete that an event is still due; the interval is what keeps N relays at N cheap queries per interval. RELAY_PRUNE set false turns it off for a project that schedules the prune itself. stop is read between the sweep's batches, so a stop request waits for at most one batch of PRUNE_BATCH_ROWS rather than for a whole backlog.

monotonic is the sweep's clock, apart from now because that one is a wall-clock reading a test sets to any value it likes for the claim, and a monotonic clock is not moved by an NTP step, which would silence or flood the sweep. It advances only when a sweep was attempted, and a failed one counts: it is not retried before the next interval.

now defaults to django.utils.timezone.now, so its readings follow USE_TZ as the columns they are compared with do; an injected clock may answer aware under either setting, and is converted to the project's form (test_a_relay_on_an_aware_clock_claims_and_delivers).

A receiver deferring with RetryAfter(seconds, counts=False) pauses its lane in this process for seconds, measured on now: the lane's unstarted rows in the batch are handed back rather than attempted - each would cost a call to learn what the first already said - and are claimable by any worker at once. While the pause lasts, a relay serving that lane claims nothing and idles, and one serving every lane leaves it out of its claims and goes on delivering the others, the rest of a mixed batch included. The pause is per process and shares nothing: every relay in the lane learns of the throttle from its own first deferral, at the cost of one call each. It ends at the first pass after seconds, so within POLL_SECONDS of it.

deliver_one

deliver_one(
    delivery_id: int,
    *,
    worker_id: str | None = None,
    claimed_by: str | None = None,
    claimed_at: str | None = None,
) -> DeliveryStatus | None

Run one delivery and record its outcome. None means this call recorded none.

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, unless the claim already leased it for this receiver. 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.

A row that is not owed is refused the same way, whoever asks: one already settled (succeeded, dead or orphaned), one that failed and is waiting out its backoff, or one whose due time has not come. Running any of them would repeat committed work, resurrect a dead letter past its budget, ignore a RetryAfter the destination sent, or do scheduled work early.

A receiver answering NotYet has its row moved to the time it named, with no attempt spent, and this returns PENDING: the one outcome that leaves the row owed and not failed.

claimed_by and claimed_at are the claim a task message carries, as the relay passed them to TaskBackend.enqueue. Given them, this call first takes the row: one conditional update, committed before anything else runs, that succeeds only while the row is still CLAIMED under exactly that claim, and moves it to worker_id - or to an id minted for this one take. A second copy of the message, a copy the queue held past its lease, and a copy for a row that has since settled all fail the take and return None without running the receiver.

A receiver deferring with RetryAfter(counts=False) is recorded here as anywhere, but pauses nothing: a pause belongs to the relay process that claimed the row's batch, and a call made directly holds no batch.

deliver_pending

deliver_pending(
    limit: int | None = None,
    *,
    worker_id: str = "deliver_pending",
    ignore_backoff: bool = False,
    lane: str | None = None,
    batch_size: int | None = None,
    now: datetime | None = None,
) -> 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.

lane delivers only that lane's rows, as run_relay(lane=...) claims them; "default" is every row no named lane takes. None, the default here, is every lane, which is what this function did before lanes existed and what drain_outbox promises. deliver_events --once passes the default lane unless told otherwise, as the relay does.

batch_size sizes each claim in place of BATCH_SIZE. It cannot be combined with limit, which is one claim of exactly that many rows under one lease: accepting both would silently ignore the batch, and claiming a hundred rows at once is what a small batch is asked for to prevent (test_a_limit_and_a_batch_size_together_are_refused).

A receiver deferring with RetryAfter(counts=False) sets its lane aside for the rest of the call: the lane's unstarted rows in the batch are handed back rather than attempted, and later claims leave the lane out, while every other lane goes on being delivered (test_a_deferral_sets_its_lane_aside_and_the_pass_delivers_the_rest).

now is the moment every claim of the call is made at, in place of the clock: what is due, which backoff has run out and which lease has lapsed are all read against it. It is how drain_outbox(now=...) moves a test forward to a receiver's due time. Only the claims read it; the attempt itself runs and is recorded on the real clock, since a receiver's acknowledgement is a fact about when it ran.

claim_batch

claim_batch(
    *,
    worker_id: str,
    now: datetime,
    lease: timedelta,
    limit: int,
    only_ids: list[int] | None = None,
    ignore_backoff: bool = False,
    lane: str | None = None,
    exclude_lanes: Collection[str] = (),
    exclude_ids: Collection[int] = (),
) -> 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. It never claims a row whose due time is still ahead of now - a receiver declared with due= - because that is not a wait the helper is skipping but the time the work exists for (test_ignoring_the_backoff_never_ignores_the_schedule).

lane claims only the rows of the receivers declared in it, read from the registry on every call; "default" claims every row no named lane takes, which includes the rows of a receiver that no longer exists; None claims every lane. The filter is applied to the whole of the owed condition, both arms, so a relay never takes over another lane's lapsed claims (test_a_lapsed_claim_in_a_named_lane_stays_out_of_the_default_lane).

exclude_lanes leaves lanes out of the claim, and is how a relay serving every lane keeps the others flowing while one is paused after a deferral. It is each lane's own condition negated, so it covers both arms as lane does - the rows a deferral handed back are lapsed claims, and exactly the ones not to take straight back (test_a_paused_lane_covers_its_lapsed_claims_too) - and excluding the default lane leaves only the named ones, or nothing where none is declared (test_a_paused_default_lane_with_no_named_lanes_claims_nothing). Empty, the claim is the one it was without it.

exclude_ids leaves those rows out, and is how a pass keeps a row it is done with from coming straight back when the pass was told a now beyond the real clock: one its receiver answered with NotYet, moved to a time that now has passed, or one handed to a task backend under a lease the real clock extended and now reads as lapsed. A pass on the real clock passes none, since neither is claimable again there until its time has really come (test_a_row_rescheduled_in_place_on_the_real_clock_is_not_excluded). Empty, it adds nothing to the claim: Django compiles a negated __in over no values to nothing at all, so the relay's claim is the one it was (test_without_named_lanes_the_default_lane_sends_the_unfiltered_claim).

backoff

backoff(attempt: int, *, base: float, cap: float, jitter: float) -> timedelta

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.

PermanentFailure

Bases: Exception

Raised by a durable receiver to say this delivery can never succeed.

The relay dead-letters the row on the attempt that raised it, instead of spending the rest of the attempt budget first. The budget is otherwise the only thing that can end a delivery, so a receiver that already knows the destination is gone - a 410 Gone, a deleted account, a revoked credential - would go on retrying for hours against something that has told it, in so many words, that it will never accept another.

An exception rather than a return value, and that is not a style choice. A receiver returns None by contract, and every mode calls it the same way, so a return convention would have to be invented for all three and would be silently ignored by two of them. Raising already means "this attempt failed"; this narrows it to "and every later one would too".

Only the relay reads it. An INLINE receiver raising it fails the caller's transaction exactly as any other exception does, and an ON_COMMIT one has it logged: neither has a row to dead-letter.

The row is recorded as DEAD with the attempt that raised it, and an on_failure hook sees DEAD, exactly as it would had the budget run out. A consumer's failure log therefore needs no second case to recognise it.

Beside deliver rather than at the package root, because that is the one module that reads it. It imports nothing, so it cannot close a cycle.

RetryAfter

Bases: Exception

Raised by a durable receiver to say when its next attempt should run.

The relay schedules the retry seconds from now instead of drawing it from the backoff curve. This is the one case where the other side has told us the answer: a 429 or 503 carrying Retry-After: 120 is a destination saying when it will be ready, and the exponential curve is a guess. Guessing against a rate limiter means arriving early and being refused again, which the curve then punishes by waiting longer than was asked.

By default it consumes an attempt. A retry that did not count would let a destination keep a row alive forever by answering 429 to every attempt, and nothing would ever declare it dead. Counting it keeps every delivery bounded by the budget it was fired with; a receiver expecting to be rate limited widens max_attempts at its declaration. On the last attempt of the budget the row is dead-lettered like any other failure, and the requested time is moot.

counts=False is a deferral instead, for a destination limiting how much it accepts rather than refusing this row: a throttle, a quota. The attempt is not counted and max_attempts is not spent; what ends the delivery is give_up_after= on the receiver, a bound in time measured from when the row became owed, past which the next deferral dead-letters it. The next attempt runs between seconds and twice that from now - never earlier than asked, spread so a throttled burst does not come back as one wave - and the relay that ran it pauses the receiver's lane for seconds and hands back the rest of its batch rather than spending a call on each row to learn what the first already said.

A receiver raising counts=False without declaring give_up_after has nothing that would ever end the delivery, so the deferral is counted like any other RetryAfter - no jitter, no pause - and a warning naming the receiver is logged once per process.

It is capped at MAX_RECEIVER_RETRY_DELAY_SECONDS, and a request past the cap is clamped with a warning naming both numbers. The cap is not BACKOFF_CAP_SECONDS, which bounds a curve: clamping a legitimate one-hour Retry-After to a backoff ceiling would hammer a destination that asked to be left alone, which is the opposite of the feature.

Everything else about a failed attempt is unchanged: the attempt is counted, the message is stored as last_error, and on_failure is called with FAILED. A deferral that does not count is recorded the same way, with the row FAILED and on_failure told the attempt number the receiver was given - which the next run is given again, since nothing was spent.

A deferral that does not count must ask for a wait above zero, since a zero would return at once and, spending nothing, would loop until give_up_after. A negative delay is refused where it is constructed. The relay would record that refusal as an ordinary failure on the ordinary curve, with a message saying why, rather than scheduling an attempt in the past.

NotYet

Bases: Exception

Raised by a durable receiver to say its work is not due until until.

The answer to a delivery that runs early. A receiver declared with due= has its due time resolved when the event fires, and the world moves on after that: an admin extends an offer from 48 hours to 72, the 48-hour row runs, finds the offer still open - and returning would succeed and leave nothing that ever expires it. Raising this instead moves the row to until and lets it run again then, so the package does not depend on the application remembering to fire a follow-up event for every change that moves a due time.

The relay writes until to the row's available_at and due_at together, sets it back to PENDING and does nothing else:

• No attempt is spent, and the next run is handed the same attempt number, because nothing failed. max_attempts cannot end it.
• It is not capped by MAX_RECEIVER_RETRY_DELAY_SECONDS, and not bounded by give_up_after. Those exist because a destination's advice about its own readiness cannot be trusted to be sane; this is the application's own declaration of when its work is due, as due= is. outbox_health() shows the row as scheduled until then.
• It is not a failure: last_error is not written and on_failure is not called.
• It sets nothing else aside. A RetryAfter(counts=False) pauses its lane because a throttle limits every row behind it; this is about one row, and the rest of the batch is delivered.

DeliveryContext.due_at on the next run is until, which is what the receiver compares against: recompute the due time from current state, and go ahead once it is no later than the context's. The comparison is if moved > max(ctx.due_at, timezone.now()): raise NotYet(moved), with the clock as a floor: a row that runs after the moved time has nothing to wait for, and comparing against ctx.due_at alone would raise a time already come, which is refused below and recorded as an ordinary failure on every retry (test_the_documented_pattern_does_the_work_of_a_row_that_runs_after_the_moved_time).

until is an aware datetime, and in the future when this is constructed. A naive one is refused under USE_TZ = True, as due= refuses one; under False it is local time, the project's own convention, and an aware one is converted to local time, so until is always in the form the row will store. A time already come is refused whatever the setting: spending no attempt, the row would be claimed straight back and answered the same way, a hot loop nothing would end. Both are refused where the exception is built, so a receiver raising either has raised an ordinary TypeError or ValueError instead, which the relay records as an ordinary failure, counted, with that message.

Not a RetryAfter spelling, although it rests on the same machinery - an outcome read at the one boundary every execution site shares, recorded without touching attempts. RetryAfter is a destination's advice, so it is counted by default, capped, jittered, bounded by give_up_after and pauses its lane; this is none of those, and a mode of that class switching each of them off would be more to learn than a class with one behaviour.

Only a durable receiver can be rescheduled: the relay reads it and nothing else does. An INLINE receiver raising it fails the caller's transaction like any other exception (test_an_inline_receiver_raising_it_fails_the_callers_change), and an ON_COMMIT one has it logged (test_an_on_commit_receiver_raising_it_has_it_logged).

notify_relay

notify_relay(
    *,
    supported: bool | None = None,
    connection: Any | None = None,
    clock: Callable[[], float] = time_module.monotonic,
) -> None

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.

Sends nothing under WAKE = "poll", and at most one per NOTIFY_COALESCE_SECONDS per database from this process. Skipping one is safe for the same reason losing one is: a single wake makes the relay claim everything due, and the poll picks up whatever the skipped one announced. clock is injected, as sleep is below, so a test moves time rather than waiting for it.

The hook around each durable attempt is a setting rather than a symbol: see AROUND_DELIVERY. It is handed the DeliveryContext below, whose receiver_key names the receiver.

Operations

cancel_deliveries

cancel_deliveries(
    event_ids: Iterable[int],
    *,
    receiver_key: str | None = None,
    delivery_ids: Iterable[int] | None = None,
) -> dict[str, int]

Withdraw the owed deliveries of these events, so they never run.

The scheduled case is the one this exists for: an offer accepted before its expiry is due, a reminder for an order already paid. A receiver that finds nothing to do and returns is still the answer when the world moved on and nobody said so; this is the answer when the application knows ahead of time, and wants neither the run nor a success recorded for work that was never done.

A delivery is withdrawn if it is unclaimed and owed: pending, which is a row waiting for its due time or for the relay, or failed, which is one waiting out its backoff - a scheduled row that came due and failed is still work, and "never runs" has to cover its retries. It becomes CANCELLED, a settled status of its own, with completed_at set and nothing else touched. Not only scheduled rows: an immediate row the relay has not reached yet is withdrawn the same way, though the relay may well get there first.

A claimed row is left alone. A worker holds it and may be running the receiver now, so settling it under the worker would record an outcome the receiver contradicts; the receiver re-checks current state when it runs, as it always did. It is counted, so the caller learns how many it could not stop. A settled row - succeeded, dead, orphaned or already cancelled - keeps its outcome: a dead letter withdrawn here would leave the dead-letter counts without anyone having read or requeued it.

Narrowed to one receiver by receiver_key, because one event usually drives an immediate reaction and a scheduled one, and only the scheduled one became moot. Narrowed to rows by delivery_ids, which intersects with event_ids rather than widening it: a stale id list cannot reach another event's work.

Returns {"cancelled": n, "in_flight": m}: the rows withdrawn, and the claimed rows in the same scope that were left to run.

Called inside a transaction on the database the package writes to, it joins it, so the cancel commits with the change that made the work moot or not at all. replay_events() reopens a cancelled row like any settled one; requeue_dead() does not touch it.

Concurrency, on a backend with row locks. Each batch locks its events first, in primary key order, the lock replay_events, requeue_dead and prune_events take: a cancel issued while a replay of the event is open waits for it and then withdraws what it reopened (test_a_cancel_issued_during_a_replay_is_not_overtaken_by_it), and two callers locking overlapping events queue rather than deadlock, which the ordering in test_the_events_are_locked_before_their_rows_are_written holds. Against the prune the lock buys nothing observable: a prune deletes only an event with nothing owed and this writes only owed rows, so the two never contend for a delivery row (test_a_prune_during_a_cancel_loses_nothing_and_takes_the_event_after passes without it). A relay's claim takes no event lock, so the write itself carries the status predicate: a claim committed first leaves the row claimed and the write skips it (test_it_will_not_cancel_a_row_claimed_after_the_lock, and on two connections test_a_cancel_waits_for_a_claim_in_flight_and_leaves_the_row_ claimed), and a write committed first leaves nothing for a claim to take.

Each condition of the write is one branch arc, so each has a test that fails without it: pending by test_unclaimed_rows_are_settled_and_never_run; failed by test_a_failed_row_awaiting_its_retry_is_cancelled_too; the event scope by test_only_the_named_events_are_touched; the receiver by test_it_can_be_narrowed_to_one_receiver; the named rows by test_it_can_be_narrowed_to_named_rows, their intersection with the events by test_a_named_row_of_an_event_not_named_is_left_alone, and an empty list meaning no rows rather than every row by test_an_empty_row_list_cancels_nothing.

prune_events

prune_events(
    older_than: timedelta | None = None,
    *,
    now: datetime | None = None,
    batch_size: int | None = None,
    limit: int | None = None,
    stop: Callable[[], bool] | None = None,
) -> int

Delete the events that are due, 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. Nothing else deletes: an event's retention is carried out here or not at all.

Due means one of three things, read off the row rather than the registry, so re-declaring or deleting an event class changes nothing for events already recorded:

• Consumed, for an event recorded under Retention.SUCCEEDED (every delivery succeeded or was cancelled) or Retention.SETTLED (every delivery terminal). No window: it goes at the first prune after it is consumed, and an event with no delivery rows at all is consumed as soon as it commits.
• Past its own window, for an event declared with a timedelta, and settled.
• Past the ordinary window, older_than or RETENTION_DAYS, and settled, for every other event - including one recorded under Retention.SUCCEEDED that a dead letter kept, so the dead letter stays inspectable for the window rather than forever. older_than replaces the ordinary window only; an event with a window of its own keeps it.

Settled means no delivery pending, failed or claimed: one still owed is work nobody recorded as lost, and deleting it would drop it. An event with no delivery rows - suppressed, or fired with no durable receivers - is settled by definition.

Deletes in batches of at most batch_size rows (default PRUNE_BATCH_ROWS), counting each event's delivery rows and the event row itself, so one transaction's lock and write volume stay bounded however the rows are spread. An event with more delivery rows than that has them deleted a batch at a time, each chunk in its own transaction, and goes with the last of them; until then it is an event with fewer deliveries, every one of them terminal. limit counts events.

Every batch records each receiver's newest success among the rows it deletes, in the same transaction, which is what lets quiet_receivers() still report a receiver whose events are all gone.

Four queries, one per policy and window, none of which reads history: the ordinary window is a range on recorded_at; each policy and the windows of their own read a partial index holding only their own live events, and check each against a partial index of unfinished delivery rows. With nothing due, the cost grows with how many events with a retention of their own are alive, and not with how much the tables have ever held. docs/retention.md has the measurement.

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.

Both are owed now, so both get due_at set to the moment of the replay. A receiver's give_up_after is measured from it, and measuring from the event's own timestamp would dead-letter a replayed month-old event on its first deferral (test_a_replayed_old_event_is_not_dead_lettered_on_its_first_deferral and test_a_row_a_replay_adds_is_owed_from_the_replay_too).

A receiver declared with due= is the exception: its due time is resolved again, through the resolver fire() uses, and both columns are written with the later of that and now. Replay is how a scheduled receiver is adopted - declare the expiry, replay the recent offers - and writing now would expire every open offer at once; an offer whose time has already passed is expired by the replay, owed from it. A due callable is called again, as a targets callable is, and raises to the operator on the same terms.

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.

A receiver declared with targets= has its callable called again, so a replay goes to the targets that exist now: a target it still returns is reopened or added exactly as a receiver is, and one it no longer returns is left as it was - neither reopened nor counted - because the callable has just said it is not owed this event. That includes a blank row written before the receiver gained targets=. A replay is a new delivery, not a re-run of an old one, and a target registered after the event was fired receiving it is the point rather than a side effect.

Calling the callable means rebuilding the event, so a fan-out receiver's replay fails loudly on a payload that no longer decodes, where a plain receiver's reopened row would dead-letter in the relay instead. Either is raised to the operator who asked; only the second waits to be found.

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. due_at resets to now for the same reason, applied to time: a receiver's give_up_after is measured from it, and a dead letter requeued a month after its event would otherwise be past a one-day bound on its first deferral (test_a_requeued_row_is_owed_from_the_requeue).

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

catalogue() -> 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.

Wildcard receivers are listed once, in their own section, and not under each event they receive - which is every event.

render_catalogue

render_catalogue(catalogue: Catalogue, *, format: str = 'markdown') -> str

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

what_listens_to(event_class: type) -> list[RegisteredReceiver]

Every receiver declared for one event, sorted by key - plus every wildcard.

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.

The wildcards are the "plus", and they are not repeated here. A receiver declared for AnyEvent receives this event as it receives every other, and listing a transport under every event answers the question two hundred times without saying anything about this one. Ask for them directly with what_listens_to(AnyEvent), which returns exactly those.

listens_for

listens_for(receiver_key: str) -> RegisteredEvent | None

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 - and a third that does not: a wildcard receiver, declared for AnyEvent, is owed every event and so names no single one. Its delivery row already says which event it was.

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 rows are not the whole record: the prune deletes them, within a sweep for an event declared to delete on consumption, and writes each receiver's newest success among what it deleted to ReceiverLastSuccess as it goes. The answer is the later of that and the live rows, so a receiver whose every event has been pruned still reports when it last ran.

The window defaults to RETENTION_DAYS, the window an ordinary event's deliveries are kept for.

outbox_health

outbox_health(*, now: datetime | None = None) -> OutboxHealth

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 and due", 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. A row whose due time is still ahead of now is not outstanding yet, so it is counted as scheduled instead, and is never what the backlog age is read from: a receiver declared with due= fourteen days out would otherwise hold a five-minute alert open for the fortnight.

Testing

drain_outbox

drain_outbox(
    limit: int | None = None,
    *,
    respect_backoff: bool = False,
    now: datetime | None = None,
) -> 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.

It never skips a receiver's due=. A row whose due time has not come is left alone, because a scheduled reaction is not a wait to be skipped but the time the work exists for: a test that makes an offer would otherwise expire it in the same breath. now is how a test moves forward instead - drain_outbox(now=placed + timedelta(hours=48)) delivers what is due by then, reading the backoff, when respected, against the same moment. The receiver still runs on the real clock; DeliveryContext.due_at is what it should compare against.

A row is attempted at most once a call when its receiver answers NotYet, and handed to a task backend at most once a call when its receiver declares site="task". A handed-off row is not counted in the result, since nothing has happened to it here, and that holds for a backend that runs the task inside the hand-off too. The pass keeps such a row out of its later claims only when it was given now: on the real clock the time a NotYet moved a row to and the lease a hand-off extended are ahead of the clock when written, so a later claim takes the row again only once that time has really come, and nothing is excluded or read back (test_a_task_row_on_the_real_clock_is_not_read_back_or_excluded).

assert_fired

assert_fired(event_class: type[E], *, times: int | None = None) -> list[E]

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

INLINE = 'inline'

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

ON_COMMIT = 'on_commit'

After commit, in the firing process, best effort. Not recoverable.

DURABLE class-attribute instance-attribute

DURABLE = 'durable'

After commit, at-least-once, retried, with a row recording the debt.

DeliveryStatus

Bases: TextChoices

Status of one delivery: an event, a durable receiver, and its target if any.

FAILED class-attribute instance-attribute

FAILED = ('failed', 'Failed')

Raised, with attempts remaining. Distinct from PENDING so that "has this ever failed" is answerable without reading the attempt count.

ORPHANED class-attribute instance-attribute

ORPHANED = ('orphaned', 'Orphaned')

Addressed to a receiver key the registry no longer has.

CANCELLED class-attribute instance-attribute

CANCELLED = ('cancelled', 'Cancelled')

Settled by cancel_deliveries before it ran: the work was known to be moot ahead of time, so it never will. Not a success, because the receiver did not run, and not a dead letter, because nothing failed.

Retention

Bases: Enum

Delete an event as soon as it has been consumed, rather than when a retention window passes.

Passed to @event(retention=...), the alternative to a timedelta. Copied onto every event row fire() records, so the fate of an event is settled when it is recorded: re-declaring or deleting the class later changes nothing for rows already written.

The values are what lands in EventRecord.delete_when. Nothing deletes until prune_events runs, so a policy is only as prompt as the schedule running it.

SUCCEEDED class-attribute instance-attribute

SUCCEEDED = 'succeeded'

Once every delivery has succeeded or been cancelled.

A dead or orphaned delivery keeps its event under the ordinary RETENTION_DAYS window, so the dead letter stays inspectable and requeue_dead still has a row to reopen. A cancelled one does not: it was withdrawn on purpose by cancel_deliveries, nothing failed and nothing is left to requeue, so with every other row succeeded every row of the event is done.

SETTLED class-attribute instance-attribute

SETTLED = 'settled'

Once every delivery is terminal, dead letters included.

A dead letter goes with its event, at the next prune, so nothing is left for requeue_dead or the dead-letter counts to find.

DeliveryContext dataclass

Delivery metadata, passed to receivers declaring takes_context.

Also to a targets= callable, at fire time, and to the AROUND_DELIVERY hook, once per durable attempt.

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.

target class-attribute instance-attribute

target: str = ''

Which target of a fan-out receiver this delivery is for.

Blank for a receiver declared without targets=, and for any row written before the receiver gained one - which is the case worth handling: adding targets= to a receiver with deliveries still owed leaves those rows blank, and they are delivered as they were written.

Last, and defaulted, so adding it does not break a consumer constructing one - this is an exported type.

delivery_id class-attribute instance-attribute

delivery_id: int | None = None

The primary key of the delivery row this attempt is for.

Set only on the context the relay or a task builds for a delivery. None where no row exists: the context targets= callables receive at fire time, which is what decides the rows, and the one INLINE and ON_COMMIT receivers receive, which never have one. Optional and last for the same reason as target.

due_at class-attribute instance-attribute

due_at: datetime | None = None

When this delivery became due.

For a receiver declared with due=, the due time its row was written with - which is what a scheduled receiver compares against: recompute the due time from current state, and if it has moved later since the row was written (the offer was extended), answer with NotYet rather than doing the work early. For any other row, when it became owed: the moment a replay or requeue reopened it, or else when its event was recorded.

None where no row exists, as for delivery_id. Optional and last for the same reason as target.

receiver_key class-attribute instance-attribute

receiver_key: str = ''

The key of the receiver this context was built for.

Set wherever a receiver is known: on every durable delivery, on the context an INLINE or ON_COMMIT receiver receives, and on the one a targets= callable is asked with. What an AROUND_DELIVERY hook labels a span, a metric or a log line by, since the hook is handed this context and not the receiver. Blank only on a context built by hand. Optional and last for the same reason as target.

DeliveryFailure dataclass

One failed attempt, handed to a receiver's on_failure hook.

Carries the delivery's identity rather than the delivery row, deliberately. The hook runs at a moment when the row has just been written by the relay, and handing over a model instance invites a second write against a row another worker may now own.

attempt instance-attribute

attempt: int

Which attempt this was, counting from one.

status instance-attribute

status: DeliveryStatus

FAILED while attempts remain, DEAD once the budget is spent. The hook is called for both, because "it failed again" and "it will not be tried again" are different things to record.

error instance-attribute

error: str

The message the relay stored, already truncated as the column is.

target class-attribute instance-attribute

target: str = ''

The target this delivery was for, blank for a receiver without targets=. Part of the delivery's identity for a fan-out receiver, since receiver_key alone no longer says which of its deliveries failed. Defaulted so adding it does not break a consumer constructing one.

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

merged(other: Scope) -> Scope

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.

wildcard_receivers class-attribute instance-attribute

wildcard_receivers: tuple[CatalogueReceiver, ...] = ()

Receivers declared for AnyEvent, which receive every event above.

A section of their own rather than a row under every event: a transport listed under all two hundred events is a catalogue nobody reads, and it says nothing about any one of them. Defaulted so adding it does not break a consumer constructing one.

CatalogueEvent dataclass

One declared event, its payload shape and everything listening to it.

migrates_older_rows class-attribute instance-attribute

migrates_older_rows: bool = False

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.

retention_seconds class-attribute instance-attribute

retention_seconds: int | None = None

The window of its own the event is declared with, in seconds, or None.

Published as fire() records it on the row rather than as the declared timedelta or Retention, so the JSON form stays plain values, and a pipeline diffing catalogues sees an event start or stop deleting early. Defaulted for the same reason as migrates_older_rows.

delete_when class-attribute instance-attribute

delete_when: str = ''

"succeeded" or "settled" for an event deleted once consumed (the Retention value), or blank. At most one of this and retention_seconds is set; with neither, the event follows RETENTION_DAYS.

CatalogueField dataclass

One field of an event payload, as the catalogue describes it.

required instance-attribute

required: bool

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

lease_seconds: int | None = None

Defaulted so adding it does not break a consumer constructing one.

targets class-attribute instance-attribute

targets: str | None = None

Where a fan-out receiver's targets callable lives, or None for a receiver that writes one delivery per event. Published because it changes what a delivery is for this receiver - one per target rather than one per event - and a reader diffing catalogues should see that change. Defaulted for the same reason as lease_seconds.

backoff_base_seconds class-attribute instance-attribute

backoff_base_seconds: float | None = None

The receiver's own retry curve, or None where it uses the setting. Defaulted, like every field after lease_seconds.

backoff_cap_seconds class-attribute instance-attribute

backoff_cap_seconds: float | None = None

The ceiling of that curve, or None where it uses the setting.

lane class-attribute instance-attribute

lane: str = 'default'

Which relay processes claim this receiver's rows: those started with --lane naming it, or, for "default", those started with none.

give_up_after_seconds class-attribute instance-attribute

give_up_after_seconds: float | None = None

How long a delivery may stay owed while it defers without counting, in seconds, or None where the receiver declares no bound. Seconds rather than the declared timedelta so the JSON stays plain numbers, as the curve does.

due_seconds class-attribute instance-attribute

due_seconds: float | None = None

For a receiver declared with due= and a delay, that delay in seconds, counted from the event's occurred_at; None otherwise. Published because it changes when the work happens, which a reader diffing catalogues should see.

due_callable class-attribute instance-attribute

due_callable: str | None = None

For a receiver declared with due= and a callable, where the callable lives, as targets is published; None otherwise. Two fields rather than one that is sometimes a number and sometimes a path, so a JSON reader never switches on a value's type. At most one of the two is set, and a receiver with neither runs as soon as its event is recorded.

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.

owed instance-attribute

owed: int

Deliveries due and not yet settled: pending, failed or claimed, with their due time come. A delivery scheduled for later is counted in scheduled instead, because it is work that has not arrived.

oldest_owed_at instance-attribute

oldest_owed_at: datetime | None

When the oldest still-owed delivery became owed, or None when nothing is owed.

That is its due_at where the row has one - the due time of a receiver declared with due=, or the moment a replay or requeue reopened it - and its event's recorded_at otherwise. Work arrives when it becomes due, so a row scheduled for a fortnight ahead does not age while it waits, and one failing after it came due ages from its due time rather than from an event three days older.

Not when it next becomes available: 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

lapsed_leases: int

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

receivers: tuple[ReceiverBacklog, ...]

Only receivers with something owed, dead or scheduled, worst backlog first.

scheduled class-attribute instance-attribute

scheduled: int = 0

Deliveries not yet settled whose due time is still ahead: the rows of a receiver declared with due= that are waiting for it. Not owed and not aged, so a backlog alert does not fire for work that is on time. Defaulted and last, so adding it does not break a consumer constructing one.

QuietReceiver dataclass

A declared receiver with nothing delivered to it inside the window.

last_succeeded_at instance-attribute

last_succeeded_at: datetime | None

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

oldest_owed_at: datetime | None

When the oldest still-owed delivery became owed. See OutboxHealth.oldest_owed_at for what that is read from.

scheduled class-attribute instance-attribute

scheduled: int = 0

Deliveries whose due time has not come yet, which owed leaves out. Defaulted and last, so adding it does not break a consumer constructing one - this is an exported type.

claimed class-attribute instance-attribute

claimed: int = 0

How many of owed a worker holds right now, the per-receiver half of OutboxHealth.claimed. A receiver whose owed rows are all claimed and none finishing is wedged, or its task queue is holding them, where the same total spread across receivers is a relay at work. Defaulted and last for the reason scheduled is.

RegisteredEvent dataclass

retention class-attribute instance-attribute

retention: timedelta | Retention | None = None

How long fire() records this event for: its own window, a :class:Retention policy, or None for RETENTION_DAYS.

Last, and defaulted, so adding it does not break a consumer constructing one - this is an exported type.

retention_columns property

retention_columns: tuple[int | None, str]

retention as the event row records it: (retention_seconds, delete_when).

Exactly one of the two is set, or neither. The schema cannot say so - the check-constraint keyword changed inside the supported Django range - so this is where it holds, for fire() and the catalogue alike.

RegisteredReceiver dataclass

on_failure class-attribute instance-attribute

on_failure: Callable[[DeliveryFailure], Awaitable[None] | None] | None = None

Called after a failed attempt has been recorded, or None.

The hook exists because a receiver cannot otherwise keep anything about its own failure: it runs inside the transaction that carries its acknowledgement, so everything it wrote is rolled back before the failure is recorded. A receiver that wants a durable log of what went wrong had no way to write one.

It runs outside that transaction, after the delivery row is updated, so what it writes survives. It must not raise: a hook that does is logged and swallowed, because the alternative is a failure path that fails.

lease_seconds class-attribute instance-attribute

lease_seconds: int | None = None

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.

targets class-attribute instance-attribute

targets: Callable[[Any, DeliveryContext], Iterable[str]] | None = None

Called at fire time to name this receiver's targets, or None.

None writes one delivery row per event, as every receiver always has. A callable writes one per string it returns, each with its own attempt count, backoff and dead-letter, and none when it returns nothing. It runs inside the caller's transaction; see receiver for what that obliges.

backoff_base_seconds class-attribute instance-attribute

backoff_base_seconds: float | None = None

None means the BACKOFF_BASE_SECONDS setting.

Read from here when an attempt fails, never copied onto the row, so a changed curve reaches deliveries already in flight. max_attempts is the opposite on purpose: it is copied, so lowering it cannot dead-letter them.

backoff_cap_seconds class-attribute instance-attribute

backoff_cap_seconds: float | None = None

None means the BACKOFF_CAP_SECONDS setting. Read like the base.

lane class-attribute instance-attribute

lane: str = DEFAULT_LANE

Which relay processes claim this receiver's rows.

Read from here at claim time, never copied onto the row, so moving a receiver to another lane moves the deliveries it is still owed with it.

give_up_after class-attribute instance-attribute

give_up_after: timedelta | None = None

How long a delivery may stay owed while deferring, or None.

What ends a delivery whose receiver defers with RetryAfter(counts=False), which spends no attempt: once the row has been owed this long - measured from due_at, or the event's recorded_at where that is NULL - the next deferral dead-letters it. None means a deferral that does not count is counted after all, because nothing else would end the delivery. Read live, like the curve.

due class-attribute instance-attribute

due: timedelta | Callable[[Any], datetime] | None = None

When this receiver's delivery of an event comes due, or None for at once.

A timedelta counts from the event's occurred_at; a callable is handed the event and returns an aware datetime. Resolved at fire time, inside the caller's transaction, and written onto the row as both its available_at and its due_at - never read again from here, so a deploy changing the delay reaches the next event, not the rows already waiting. replay_events resolves it afresh.

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.

It may also deliver one twice, or late, which is what acks_late and a broker's visibility timeout do. The message carries the claim the row was handed off under, and the task takes the row under exactly that claim before running anything, so a second copy or a copy the queue held past its lease finds the claim gone and does nothing.

enqueue

enqueue(delivery_id: int, claimed_by: str, claimed_at: str) -> None

Arrange for the delivery to run somewhere, carrying its claim.

The worker must call deliver_one(delivery_id, claimed_by=claimed_by, claimed_at=claimed_at) with the three values exactly as given. claimed_at is an ISO 8601 string rather than a datetime, so all three are JSON and any queue can carry them unchanged. The relay passes the claim by keyword.

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.

Task backends

Not exported from the package root. Each is named by its dotted path in TASK_BACKEND, so the framework it wraps is imported only by a project that chose it.

DjangoTasksBackend

Hands deliveries to Django's Tasks framework.

Django Tasks rather than Celery first: django.tasks is in core from 6.0 and the django-tasks backport covers 4.2 through 6.0, so it spans this package's whole supported range with no broker to install.

The import is lazy because the framework is not a dependency - a consumer using the relay directly should not have to have it.

CeleryBackend

Hands deliveries to Celery.

For a project whose workers already run Celery. A project without one is better served by DjangoTasksBackend, which needs no broker.

Redelivery is safe: acks_late and a broker's visibility timeout both hand a worker a message whose task already ran, and the claim the message carries is what turns that copy into a no-op. Recovery from a worker that died mid-delivery comes from the row's lease, not from the redelivery: the second copy finds the row taken by the worker that died and leaves it for the relay to reclaim.