django-domain-events

A Django domain-event log with in-process fan-out.

fire() records a typed event to a database table inside the caller's transaction; a relay delivers it to registered receivers afterwards, at-least-once, with per-receiver retry and dead-lettering. The event exists if and only if the change committed.

This is not a signals replacement. A database write per event rules out chatty notification use, and buys three things signals cannot give you: a crash story, durable attribution for who caused what, and an event log you can query.

Install

pip install django-domain-events

Add it to INSTALLED_APPS and migrate:

INSTALLED_APPS = [..., "django.contrib.auth", "django_domain_events"]

django.contrib.auth is required: the event row carries a nullable foreign key to AUTH_USER_MODEL so attribution survives, and the migration depends on it.

python manage.py migrate

Nested event payloads need the decode half of the codec:

pip install "django-domain-events[dacite]"

Quickstart

Declare an event and something that listens for it:

# orders/events.py
from dataclasses import dataclass

from django_domain_events import DURABLE, event, receiver


@event
@dataclass(frozen=True, slots=True)
class OrderPlaced:
    order_id: int
    total_cents: int


@receiver(OrderPlaced, mode=DURABLE)
def reserve_stock(evt: OrderPlaced) -> None: ...

Fire it inside the transaction that makes the change:

with transaction.atomic():
    order = Order.objects.create(...)
    fire(OrderPlaced(order_id=order.id, total_cents=order.total_cents))

The event row and one delivery row per durable receiver are written in that same transaction. Run the relay to deliver what is owed:

python manage.py deliver_events          # claim and deliver continuously
python manage.py deliver_events --once   # one pass, for cron or CI

The relay claims with SELECT ... FOR UPDATE SKIP LOCKED under a lease, so you can run as many as you like: two workers never take the same row, and one that dies without acknowledging has its rows reclaimed when the lease lapses. Failed deliveries retry with exponential backoff and full jitter, then dead-letter.

Add eager=True to a receiver to also attempt it immediately after commit, in the firing process, with the relay as the fallback.

In tests, drain_outbox() runs the real delivery path to completion, and assert_fired(OrderPlaced, times=1) reads the log rather than a mock.

Attribution

with attributed(actor=request.user, source="checkout"):
    with transaction.atomic():
        order = Order.objects.create(...)
        fire(OrderPlaced(order_id=order.id, total_cents=order.total_cents))

Every event fired inside the block records who caused it, in what scope, and which chain it belongs to. The scope is captured at fire time and read back off the row, so a delivery running hours later in another process still knows.

Suppress without losing the record:

with suppressed(OrderPlaced, reason="historical import"):
    importer.run()  # rows written and marked, no deliveries

Delivery modes

Two independent knobs, not one enum. Timing is what a receiver promises about the transaction; where its code runs is a separate question, and only meaningful for DURABLE.

For a receiver that touches only this database, the work and the acknowledgement commit together, so delivery is effectively once: the duplicate an at-least-once system owes you cannot be observed. Receivers with side effects outside the database are at-least-once, as promised.

Status

Early development. The API is not stable and the package is not yet usable; see the changelog for what has landed.