import uuid
from django.db import models
from django.utils import timezone
class OutboxManager(models.Manager):
def available(self, limit=100):
now = timezone.now()
return (
self.filter(status=OutboxMessage.PENDING, next_retry_at__lte=now)
.order_by("created_at")[:limit]
)
class OutboxMessage(models.Model):
PENDING, SENT, FAILED = "pending", "sent", "failed"
STATUS_CHOICES = [(PENDING, "Pending"), (SENT, "Sent"), (FAILED, "Failed")]
id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
topic = models.CharField(max_length=200)
payload = models.JSONField()
dedupe_key = models.CharField(max_length=255, unique=True)
status = models.CharField(max_length=16, choices=STATUS_CHOICES, default=PENDING)
attempts = models.PositiveIntegerField(default=0)
max_attempts = models.PositiveIntegerField(default=8)
last_error = models.TextField(blank=True, default="")
next_retry_at = models.DateTimeField(default=timezone.now)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)
objects = OutboxManager()
class Meta:
indexes = [
models.Index(fields=["status", "next_retry_at"]),
]
def __str__(self):
return "{}:{} ({})".format(self.topic, self.dedupe_key, self.status)
import logging
from datetime import timedelta
from django.db import transaction
from django.utils import timezone
from .models import OutboxMessage
logger = logging.getLogger(__name__)
MAX_BACKOFF_SECONDS = 3600
def _schedule_retry(message, error):
message.attempts += 1
message.last_error = str(error)[:2000]
if message.attempts >= message.max_attempts:
message.status = OutboxMessage.FAILED
logger.error("Outbox message %s exhausted retries", message.id)
else:
delay = min(2 ** message.attempts, MAX_BACKOFF_SECONDS)
message.next_retry_at = timezone.now() + timedelta(seconds=delay)
message.save(update_fields=["attempts", "last_error", "status", "next_retry_at", "updated_at"])
def flush_once(publish, batch_size=100):
sent = 0
with transaction.atomic():
rows = list(
OutboxMessage.objects.available(limit=batch_size)
.select_for_update(skip_locked=True)
)
for message in rows:
try:
publish(message.topic, message.payload, key=message.dedupe_key)
except Exception as exc: # broker/network failures are expected
_schedule_retry(message, exc)
continue
message.status = OutboxMessage.SENT
message.save(update_fields=["status", "updated_at"])
sent += 1
return sent
from celery import shared_task
from django.db import transaction
from .flusher import flush_once
from .models import OutboxMessage
from .publisher import kafka_publish
@shared_task(bind=True, max_retries=3, default_retry_delay=30)
def flush_outbox(self, batch_size=100):
try:
return flush_once(kafka_publish, batch_size=batch_size)
except Exception as exc:
raise self.retry(exc=exc)
def enqueue_message(topic, payload, dedupe_key):
with transaction.atomic():
obj, created = OutboxMessage.objects.get_or_create(
dedupe_key=dedupe_key,
defaults={"topic": topic, "payload": payload},
)
return obj, created
# celery beat schedule (settings.py)
CELERY_BEAT_SCHEDULE = {
"flush-outbox-every-5s": {
"task": "outbox.tasks.flush_outbox",
"schedule": 5.0,
"kwargs": {"batch_size": 200},
},
}
The transactional outbox pattern solves a stubborn distributed-systems problem: how to atomically commit a database change and publish a message about it. Writing to the database and then calling a broker in the same request is not atomic — a crash between the two leaves the system inconsistent. Instead the message is written to an outbox table inside the same transaction as the business change, and a separate process later reads pending rows and publishes them. This snippet builds that flusher end to end in Django.
In models.py, OutboxMessage is the durable queue row. Each record carries a topic, a JSON payload, and a dedupe_key that is unique, giving idempotency: producers can safely retry writing the same logical event without creating duplicates. The status, attempts, and next_retry_at columns track delivery state, and available() is a manager method that selects rows that are PENDING and due, ordered oldest-first so delivery is roughly FIFO.
In flusher.py, flush_once() is the heart of the worker. It opens a transaction and uses select_for_update(skip_locked=True) so multiple flusher processes can run concurrently without stepping on each other — Postgres hands each worker a disjoint batch and skips rows another worker already locked. Each message is passed to a publish callable; on success the row is marked SENT, and on failure _schedule_retry computes an exponential backoff (2 ** attempts seconds, capped) and pushes next_retry_at into the future, or marks the row FAILED once max_attempts is exhausted. Because publishing happens outside the row's own commit isn't possible here, the code keeps the DB transaction short and tolerates at-least-once delivery — the consumer must dedupe on dedupe_key.
In tasks.py, flush_outbox is a Celery task wrapping flush_once, and Celery Beat schedules it on a fixed interval so pending messages drain on a timer even when no new writes arrive. enqueue_message shows the producer side: it writes the outbox row inside transaction.atomic() alongside whatever business logic committed the change. The main trade-offs are added write amplification and eventual (not instant) delivery, in exchange for never losing a message to a mid-flight crash. This approach is worth reaching for whenever a service must reliably emit events to Kafka, a webhook, or another service.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
from django.urls import path
from . import views
app_name = 'blog'
urlpatterns = [Django URL namespacing and reverse lookups
import graphene
from graphene_django import DjangoObjectType
from blog.models import Post, Comment
class PostType(DjangoObjectType):Django GraphQL with Graphene
Share this code
Here's the card — post it anywhere.
