--- title: Database Change Notifications slug: database-notifications level: advanced order: 11 --- # Database Change Notifications Subscribe LiveViews to PostgreSQL `LISTEN/NOTIFY` so database changes push real-time updates to connected users — without writing any explicit pub/sub wiring. Shipped in **v0.5.0** (`djust.db.notify_on_save`, `djust.db.send_pg_notify`, `NotificationMixin.listen`). ## The 30-second version ```python from django.db import models from djust import LiveView from djust.db import notify_on_save @notify_on_save # default channel: "shop_order" class Order(models.Model): status = models.CharField(max_length=20) class OrderDashboard(LiveView): template_name = "dashboard.html" def mount(self, request, **kwargs): self.orders = list(Order.objects.filter(status="pending")) self.listen("shop_order") def handle_info(self, message): if message["type"] == "db_notify": self.orders = list(Order.objects.filter(status="pending")) ``` Create / update / delete an `Order` from anywhere — a Django admin, a Celery task, a management command, even a `psql` shell — and every user viewing `OrderDashboard` gets fresh data, usually within a few milliseconds. Delivery is best-effort; see [Dropped notifications](#dropped-notifications-under-contention). ## How it works ``` ┌─────────────┐ post_save ┌────────────────┐ NOTIFY ┌─────────────┐ │ Model.save()├───────────────►│send_pg_notify()├──────────►│ Postgres │ └─────────────┘ (signal) └────────────────┘ SQL │ LISTEN ch. │ └──────┬──────┘ │ ▼ group_send ┌─────────────────────────────────┐ ┌─────────────┐◄────── ───────────────── │ PostgresNotifyListener │ │ LiveView's │ "djust_db_notify_X" │ (async task, dedicated conn) │ │ handle_info │ └─────────────────────────────────┘ └─────────────┘ ``` 1. `@notify_on_save` wires Django `post_save` / `post_delete` signals so every `save()` and `delete()` emits `NOTIFY , ''`. 2. A **process-wide** `PostgresNotifyListener` runs one dedicated `psycopg.AsyncConnection` and does `async for notify in conn.notifies():`. On each NOTIFY it calls `channel_layer.group_send("djust_db_notify_", ...)`. 3. After `mount()` returns, the WebSocket consumer joins the Channels group for every channel the view listens on. A `self.listen()` call in a later event handler joins at the end of that event. (Before 1.2.1, only channels declared at class level were joined; #2962.) 4. The consumer's `db_notify` handler calls `handle_info(message)` and re-renders — VDOM patches stream down to the browser. The only code you write is the decorator, the channel subscription, and `handle_info()`. ## `@notify_on_save` ```python from djust.db import notify_on_save # Default channel: "{app_label}_{model_name}" @notify_on_save class Order(models.Model): ... # Explicit channel (keyword) @notify_on_save(channel="orders") class Order(models.Model): ... # Explicit channel (positional shorthand) @notify_on_save("orders") class Order(models.Model): ... ``` **Payload shape:** ```json {"pk": 42, "event": "save", "model": "shop.Order"} ``` Minimal by design. Postgres caps NOTIFY payloads at **8000 bytes**, and djust is stricter: `send_pg_notify` logs a warning above 4 KB and drops (with an ERROR log) any payload over 7,500 bytes. Receivers re-fetch full state via the ORM when they need it. **Channel name rules:** `^[a-z_][a-z0-9_]{0,62}$`. Uppercase, hyphens, dots, and quotes are rejected at decorator-registration time. This is security-critical — Postgres `NOTIFY` doesn't accept bind parameters for the channel name, so the regex is the only defense against SQL injection. ## `self.listen(channel)` ```python def mount(self, request, **kwargs): self.listen("orders") self.listen("users") # subscribe to multiple channels self.listen("orders") # duplicate subscriptions are idempotent ``` Raises `ValueError` for bad channel names. On a non-PostgreSQL backend, or without psycopg 3.2+, `self.listen()` does **not** raise. The background listener logs a `pg listener disabled (permanent failure)` warning and no notifications are delivered. The decorator `@notify_on_save` itself degrades gracefully — it becomes a no-op with a debug log — so the same model code works in sqlite test suites. ## `handle_info(message)` ```python def handle_info(self, message): if message["type"] == "db_notify": channel = message["channel"] # e.g. "orders" payload = message["payload"] # {"pk": 42, "event": "save", ...} if channel == "orders": self.orders = list(Order.objects.filter(status="pending")) elif channel == "users": self.user_count = User.objects.count() ``` Default implementation is a no-op. Override to react. Re-rendering happens automatically after `handle_info` returns — the consumer calls `render_with_diff` and pushes VDOM patches. Setting `self._skip_render = True` inside `handle_info` suppresses the render (useful when the notification doesn't affect visible state). ## Firing NOTIFYs from other places Anywhere you can reach a Django DB connection, you can broadcast: ```python # Celery task from celery import shared_task from djust.db import send_pg_notify @shared_task def nightly_report_ready(report_id): send_pg_notify("reports", {"pk": report_id, "event": "generated"}) ``` ```python # Management command from django.core.management.base import BaseCommand from djust.db import send_pg_notify class Command(BaseCommand): def handle(self, *args, **opts): send_pg_notify("system", {"event": "maintenance_done"}) ``` ```sql -- psql / DB trigger NOTIFY orders, '{"pk": 42, "event": "save"}'; ``` `send_pg_notify` is a no-op on non-Postgres backends (debug-logged) — your tests on SQLite don't need conditional imports. ## Common patterns ### Reactive dashboards Every connected admin sees order status changes in real time — no polling, no manual broadcasting. ```python @notify_on_save(channel="orders") class Order(models.Model): status = models.CharField(max_length=20) class OrderDashboard(LoginRequiredMixin, LiveView): template_name = "admin/orders.html" def mount(self, request, **kwargs): self.listen("orders") self._refresh() def handle_info(self, message): self._refresh() def _refresh(self): self.by_status = { status: Order.objects.filter(status=status).count() for status in ("pending", "paid", "shipped") } ``` ### Collaborative editing ```python @notify_on_save(channel="document") class Document(models.Model): body = models.TextField() class DocumentView(LiveView): def mount(self, request, doc_id, **kwargs): self._doc_id = doc_id self.listen("document") self.doc = Document.objects.get(pk=doc_id) def handle_info(self, message): # The payload pk is a JSON number; URL kwargs are strings unless # the route uses , so compare as strings. if str(message["payload"].get("pk")) == str(self._doc_id): self.doc.refresh_from_db() ``` ### Admin → user broadcast A support agent changes an order's status in the Django admin; the customer's page updates instantly with no extra plumbing. ```python @notify_on_save(channel="orders") class Order(models.Model): ... class CustomerOrderView(LiveView): def mount(self, request, order_id, **kwargs): self._order_id = order_id self.listen("orders") self.order = Order.objects.get(pk=order_id) def handle_info(self, message): if str(message["payload"].get("pk")) == str(self._order_id): self.order.refresh_from_db() ``` ## Limitations & gotchas ### Missed notifications across disconnects PostgreSQL discards queued NOTIFYs when the `LISTEN` connection drops. If the listener's TCP connection to the database fails, the listener automatically reconnects (1-second backoff) and re-issues LISTEN for every channel — but **notifications emitted during the drop window are lost**. Recovery is automatic for the WebSocket side: `mount()` re-runs on WS reconnect, which re-fetches state from the DB. For server-side drops in the listener connection, any state that changed while the listener was down gets picked up on the next NOTIFY after reconnect, but intervening changes are silent. **Mitigation:** if your dashboard must never miss an update, combine NOTIFY with a periodic `handle_tick()` that refreshes from the DB (set `tick_interval = 30_000` or similar on the view; `handle_tick()` only runs when `tick_interval` is set). NOTIFY handles "instant" updates; `handle_tick()` serves as a catch-up for any missed events. ### Dropped notifications under contention Delivery is best-effort. If the view is handling a user event, or its render lock is held for more than 100 ms, the notification is dropped (debug-logged) and not queued. Under bursty notification streams some re-renders are skipped. Pair NOTIFY with a periodic refresh (above) if every change must render. ### 8000-byte payload cap Postgres's `NOTIFY` payload limit is 8000 bytes. djust warns above 4 KB and drops (with an ERROR log) any payload over 7,500 bytes, below Postgres's cap. Keep payloads minimal. Full row snapshots often exceed that; re-fetching via the ORM is the intended pattern. ### PostgreSQL only `LISTEN/NOTIFY` is Postgres-specific. SQLite, MySQL, and Oracle are not supported. `@notify_on_save` is a silent no-op on non-Postgres backends so you can develop on SQLite without conditional imports. `self.listen()` does not raise either: the background listener logs a `pg listener disabled (permanent failure)` warning and delivers nothing. ### One listener per process Each Django process that runs WebSocket consumers starts its own `PostgresNotifyListener`. This scales linearly with workers — fine for typical deployments (2-8 Channels workers), but if you run hundreds of processes you'll hit Postgres's `max_connections` ceiling. The listener uses a **dedicated** connection (not the Django connection pool), so pgbouncer transaction pooling doesn't interfere. ### Connection pooler caveats `LISTEN/NOTIFY` requires session-level connections. If you're behind pgbouncer in `transaction` mode, point the listener directly at Postgres (bypass the pooler) or use `session` mode for the djust listener connection. By default the listener reads DSN fields from `settings.DATABASES ["default"]`. To isolate the long-lived `LISTEN` connection from your request-path connection pool entirely, set **`DJUST_NOTIFY_DATABASE_URL`** — a `DATABASE_URL`-style override that takes precedence over `DATABASES["default"]` for the listener connection only: ```python # settings.py DJUST_NOTIFY_DATABASE_URL = "postgres://user:pass@direct-pg.internal:5432/appdb" ``` It can also be supplied via an environment variable of the same name (handy on PaaS): ```bash export DJUST_NOTIFY_DATABASE_URL="postgres://user:pass@direct-pg.internal:5432/appdb" ``` Point this at a direct (non-pooler) endpoint so the session-mode `LISTEN` connection can't saturate a shared transaction-pool. The setting is fully backwards-compatible — when unset, the listener uses `DATABASES["default"]` exactly as before. The postgres-only check still applies: a non-postgresql URL scheme raises `DatabaseNotificationNotSupported` inside the listener task, which logs it as a permanent failure and stops. The override URL (and its embedded password) is never logged. #### Connection query parameters (TLS, unix sockets) The override URL may carry a query string with a **known-safe allowlist** of libpq connection parameters — useful for requiring TLS on a direct endpoint or pointing the listener at a unix socket: | Query param | Example | Purpose | |--------------------|----------------------------------|----------------------------------| | `sslmode` | `?sslmode=require` | Require/verify TLS | | `sslrootcert` | `?sslrootcert=/etc/ssl/ca.pem` | CA bundle for `verify-full` | | `sslcert` | `?sslcert=/etc/ssl/client.pem` | Client cert (mTLS) | | `sslkey` | `?sslkey=/etc/ssl/client.key` | Client key (mTLS) | | `host` | `?host=/var/run/postgresql` | Unix-socket directory | | `application_name` | `?application_name=djust-listen` | Label the connection in `pg_stat`| | `connect_timeout` | `?connect_timeout=10` | Connect timeout (seconds) | ```python # Require TLS on a direct endpoint: DJUST_NOTIFY_DATABASE_URL = "postgres://user:pass@direct-pg.internal:5432/appdb?sslmode=require" # Connect over a unix socket (the URL host is treated as an ignored placeholder): DJUST_NOTIFY_DATABASE_URL = "postgres://user:pass@placeholder/appdb?host=/var/run/postgresql" ``` **`host` precedence.** When a `?host=` query item is present it **replaces** the URL netloc host, so the resulting connection has exactly one host value. This is the deterministic unix-socket behavior — the netloc host (`placeholder` above) is ignored. Any **unknown** query key is silently dropped: a query string can never override the URL-derived `user` / `password` / `dbname` (those keys are not in the allowlist), so a stray `?password=…` cannot hijack the credentials. ### Security: untrusted NOTIFY sources If Postgres is shared with other applications, anyone with DB access can emit `NOTIFY orders, '{"pk": 999999, "event": "delete"}'`. Always validate payloads in `handle_info` — don't blindly trust `pk` values or perform destructive actions based on channel messages. Treat NOTIFYs as "something changed, re-fetch via ORM" hints rather than authoritative state. ## Testing Unit tests can mock the listener: ```python from unittest.mock import patch, AsyncMock with patch("djust.db.notifications.PostgresNotifyListener") as mock: mock.instance.return_value.ensure_listening = AsyncMock() view.mount(request) assert view._listen_channels == {"orders"} ``` Integration tests that want a live round trip can use the pattern from `tests/integration/test_pg_notify_roundtrip.py`: ```python import pytest from django.conf import settings pytestmark = pytest.mark.skipif( not settings.DATABASES["default"]["ENGINE"].endswith("postgresql"), reason="needs postgresql", ) ``` ## Related APIs - `djust.push_to_view` — lower-level broadcast that skips the DB and goes straight through Channels. Use when you have direct access to the process state. - `djust.presence` — user-presence tracking via Channels groups. - `LiveView.handle_tick()` — periodic polling as a NOTIFY fallback.