How to Add Pub/Sub to a Django App
Publishing events from Django views and handling them in subscribers with Encore, Django 6's Tasks framework, Celery or Django signals, with idempotent handlers, transactions and CSRF covered.
Django views often trigger work that does not belong in the request, such as receipts, partner webhooks or search indexing. Moving that work out brings a broker, workers, retries and duplicate deliveries, plus the question of what happens when the transaction rolls back.
This guide adds Pub/Sub to a Django order view with Encore, Django 6's Tasks framework, Celery and Django signals. Each version publishes an order-created event to a receipt subscriber.
What you need to decide:
- Whether the work must survive a restart or a deploy.
- Whether one event has one consumer or several.
- How subscribers handle a duplicate delivery.
- How the event relates to the database transaction that created the order.
Pub/Sub options for Django at a glance
Encore: Pub/Sub for your existing Django project
Encore combines topics and subscriptions in your Django code with the broker behind them, running locally under encore run and as SNS and SQS on AWS or Pub/Sub on GCP in your own account after deploy. It is an infrastructure SDK added to the project, so models, the admin and Django's own migrations stay as they are.
The database and topic are declared in orders/resources.py. Django manages the schema, so the database has no migrations directory:
from encore import pubsub
from encore.storage import postgres
# Django manages the schema through its own migrations.
db = postgres.Database("shop")
# Published after an order is saved. Delivery is at least once.
order_created = pubsub.Topic("order-created")
settings.py builds DATABASES from the declared database's connection string, and the secret key comes from the environment:
import os
from psycopg.conninfo import conninfo_to_dict
from orders.resources import db
SECRET_KEY = os.environ["DJANGO_SECRET_KEY"]
connection = conninfo_to_dict(db.connection_string)
DATABASES = {
"default": {
"ENGINE": "django.db.backends.postgresql",
"NAME": connection.pop("dbname"),
"USER": connection.pop("user", ""),
"PASSWORD": connection.pop("password", ""),
"HOST": connection.pop("host", ""),
"PORT": connection.pop("port", ""),
"OPTIONS": connection,
}
}
A ReceiptSent model with a unique order_id records which orders already got a receipt, which is how the subscriber stays idempotent:
from django.db import models
class Order(models.Model):
email = models.EmailField()
total_cents = models.PositiveIntegerField()
created_at = models.DateTimeField(auto_now_add=True)
class ReceiptSent(models.Model):
# One row per order that already got its receipt.
order_id = models.BigIntegerField(unique=True)
sent_at = models.DateTimeField(auto_now_add=True)
The subscriber uses Django's async ORM methods, skips orders that already have a receipt and retries up to five times on failure:
import logging
from encore import pubsub
from .models import ReceiptSent
from .resources import order_created
logger = logging.getLogger(__name__)
async def send_receipt(msg: pubsub.Message[dict]) -> None:
order_id = msg.data["order_id"]
# Delivery is at least once, so skip orders that already got a receipt.
if await ReceiptSent.objects.filter(order_id=order_id).aexists():
return
logger.info("sending receipt for order %s to %s", order_id, msg.data["email"])
await ReceiptSent.objects.aget_or_create(order_id=order_id)
receipts = pubsub.Subscription[dict](
order_created, "send-receipt", handler=send_receipt, max_retries=5
)
The app config imports the subscribers module in ready(), so the subscription is registered before Encore starts it:
from django.apps import AppConfig
class OrdersConfig(AppConfig):
default_auto_field = "django.db.models.BigAutoField"
name = "orders"
def ready(self) -> None:
# Import the subscription so encore.start() can start it.
from . import subscribers # noqa: F401
The view saves the order and publishes the event. It is a stateless JSON endpoint without cookie authentication, which is the one case where exempting it from CSRF is correct:
import json
from django.http import HttpRequest, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from .models import Order, ReceiptSent
from .resources import order_created
@csrf_exempt # Stateless JSON API example without cookie auth.
@require_POST
async def create_order(request: HttpRequest) -> JsonResponse:
body = json.loads(request.body)
order = await Order.objects.acreate(
email=body["email"], total_cents=body["total_cents"]
)
# Published after the row is saved.
await order_created.publish({"order_id": order.pk, "email": order.email})
return JsonResponse({"id": order.pk}, status=201)
async def receipts(request: HttpRequest) -> JsonResponse:
ids = [r.order_id async for r in ReceiptSent.objects.order_by("order_id")]
return JsonResponse({"receipts_sent": ids})
Django's ASGI handler does not process lifespan events, so a small main.py wraps it, calls encore.start() at startup to run the subscriptions on Uvicorn's event loop and serves everything else through Django:
"""Serve Django over ASGI with Uvicorn and start Encore subscriptions."""
import logging
import os
import encore
import uvicorn
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
from config.asgi import application as django_app # noqa: E402
async def app(scope, receive, send):
# Django's ASGI handler does not handle lifespan events, so this wrapper
# starts Encore's subscriptions on Uvicorn's event loop at startup.
if scope["type"] == "lifespan":
while True:
message = await receive()
if message["type"] == "lifespan.startup":
await encore.start()
await send({"type": "lifespan.startup.complete"})
elif message["type"] == "lifespan.shutdown":
await send({"type": "lifespan.shutdown.complete"})
return
else:
await django_app(scope, receive, send)
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
host, _, port = os.environ.get("ENCORE_LISTEN_ADDR", "127.0.0.1:8000").rpartition(":")
uvicorn.run(app, host=host, port=int(port))
Django migrations run through encore exec so they reach the same database, and encore run starts Postgres locally and delivers messages:
Every request and message delivery gets a trace in the local logs, and tracing follows an order from the view into the subscriber. On deploy, Encore creates an SNS topic and an SQS queue for the subscription on AWS, or a Pub/Sub topic and subscription on GCP, with the IAM permissions for each, in your own account.
When an agent adds a consumer such as a partner webhook, it adds one subscription and checks the delivery in the local trace, and the cloud resources follow from that code on the next deploy.
Good to know
Publishing after the insert means a crash between the two can leave an order without its event. When that matters, write the event to an outbox table in the same transaction and publish from it, the transactional outbox pattern. Views that use session cookies keep Django's CSRF middleware and read the token from the csrftoken cookie in the client.
Django's Tasks framework
Django 6.0 added a Tasks framework with a @task decorator and enqueue(), and an async aenqueue() for async views. The documentation recommends transaction.on_commit so a worker never sees an order that has not been committed:
from functools import partial
from django.db import transaction
from django.tasks import task
from .models import Order
@task
def send_receipt(order_id: int, email: str) -> None:
print(f"sending receipt for order {order_id} to {email}")
def create_order(email: str, total_cents: int) -> int:
with transaction.atomic():
order = Order.objects.create(email=email, total_cents=total_cents)
transaction.on_commit(
partial(send_receipt.enqueue, order_id=order.pk, email=order.email)
)
return order.pk
Good to know
Django ships an immediate backend and a dummy backend for development and tests, and leaves execution to a third-party backend with its own worker. Arguments must survive a JSON round trip, so pass IDs rather than model instances. It is a task API rather than Pub/Sub, so each consumer is its own task.
Celery
Celery remains the most common choice in Django projects, with Redis or RabbitMQ as the broker:
from celery import shared_task
from django.db import transaction
@shared_task(acks_late=True, autoretry_for=(Exception,), retry_backoff=True, max_retries=5)
def send_receipt(order_id: int, email: str) -> None:
print(f"sending receipt for order {order_id} to {email}")
def order_saved(order) -> None:
transaction.on_commit(lambda: send_receipt.delay(order.pk, order.email))
Good to know
You run the broker and at least one celery worker process in every environment, including locally. .delay() is a blocking call, so in an async view wrap it with sync_to_async or call it from on_commit, which Django runs after the transaction commits.
Django signals
Signals connect receivers to events like post_save inside one process:
from django.db.models.signals import post_save
from django.dispatch import receiver
from .models import Order
@receiver(post_save, sender=Order)
def order_created(sender, instance: Order, created: bool, **kwargs) -> None:
if created:
print(f"order {instance.pk} created")
Good to know
Receivers run synchronously inside the request, so they add latency and an exception in one fails the save. Nothing is persisted or retried, which makes signals a tool for in-process decoupling rather than background work.
How to choose
- Encore if you want topics with several subscribers that run locally and deploy to SNS and SQS or Pub/Sub in your account, without running a broker yourself.
- Django's Tasks framework if you want a standard Django task API and are ready to pick and run a backend.
- Celery if the project already runs it and each event has one consumer.
- Signals for in-process hooks that can fail with the request.
Python support in Encore is in beta, and you can sign up for access to try it with your Django project.
Related guides
Frequently asked questions
How do I add Pub/Sub to a Django app?
Declare a topic, publish an event from the view once the row is saved, and handle it in subscribers that are safe to run twice. With Encore, declare order_created = pubsub.Topic("order-created"), publish with await order_created.publish({"order_id": order.pk}), and register pubsub.Subscription[dict](order_created, "send-receipt", handler=send_receipt, max_retries=5). encore run delivers messages locally, and on deploy the topic becomes SNS and SQS on AWS or Pub/Sub on GCP.
Are Django signals a message queue?
No. Signals call receivers synchronously in the same process during the request, so a slow or failing receiver slows or fails the request, and nothing is retried. They are useful for decoupling code inside one process, not for background work.
Does Django 6 have a built-in task queue?
Django 6.0 added a Tasks framework for defining and enqueueing background work, but its built-in backends are for development and testing. Production needs a third-party backend that supplies a durable queue and a worker process.
How do I avoid processing a Django event twice?
Record which events a subscriber has handled under a unique key, such as the order ID with a unique constraint, and skip the work when the record exists. Brokers like SQS and Google Cloud Pub/Sub deliver at least once, so duplicates are expected.
Deploy to your cloud on AWS and Google Cloud
without managing infrastructure
Encore lets developers and agents define application infrastructure directly in code, then automatically provisions it from local development to production in your AWS or GCP account.
