Coverage for object_streams/triggers.py: 100%
30 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-02 17:07 +0000
1"""Trigger-based outbox capture.
3Producers written in Python run after a write and, for the ``enqueue_*`` variants,
4after the transaction commits. A trigger writes the outbox row inside the same
5transaction as the write that caused it, so an event cannot be lost between commit
6and callback, cannot be forgotten at a new write site, and cannot be faked by
7application code sending a signal by hand.
9``pg_notify`` is transactional: a notification is delivered when the transaction
10commits and discarded when it rolls back. Doing the insert and the notify in one
11trigger is therefore both durable and rollback safe.
13This module requires ``django-pgtrigger``. Install the ``triggers`` extra.
15Declare capture in model state so migrations pick it up, either on a model you own:
17```python
18class Order(models.Model):
19 class Meta:
20 triggers = [ObjectStreamTrigger(name="order_stream")]
21```
23or, for a model you do not own, on a proxy in your own app so the migration lands
24in your app rather than in the one that defines the model:
26```python
27class OrderStream(ThirdPartyOrder):
28 class Meta:
29 proxy = True
30 triggers = [ObjectStreamTrigger(name="order_stream")]
31```
33Run ``makemigrations`` after declaring one. A declared trigger that was never
34migrated is not installed, and capture silently does nothing.
35"""
37from __future__ import annotations
39import pgtrigger
40from django.db import models
42from object_streams.postgres import DEFAULT_NOTIFY_CHANNEL
43from object_streams.postgres import validate_notify_channel
46__all__ = ("NOTIFY_CHANNEL_SETTING", "ObjectStreamTrigger")
49# Read at write time from a database setting rather than baked into the trigger body,
50# so changing the channel does not require a migration:
51# ALTER DATABASE mydb SET object_streams.notify_channel = 'my_channel';
52NOTIFY_CHANNEL_SETTING = "object_streams.notify_channel"
54_FUNC = """
55 IF (TG_OP = 'DELETE') THEN
56 subject_row := to_jsonb(OLD);
57 subject_op := 'deleted';
58 ELSIF (TG_OP = 'INSERT') THEN
59 subject_row := to_jsonb(NEW);
60 subject_op := 'created';
61 ELSE
62 subject_row := to_jsonb(NEW);
63 subject_op := 'updated';
64 END IF;
66 SELECT id INTO subject_type
67 FROM django_content_type
68 WHERE app_label = '{subject_meta.app_label}' AND model = '{subject_meta.model_name}';
70 IF subject_type IS NULL THEN
71 RETURN NULL;
72 END IF;
74 IF (TG_OP = 'UPDATE') THEN
75 SELECT coalesce(jsonb_agg(field_name ORDER BY field_name), '[]'::jsonb)
76 INTO changed
77 FROM jsonb_object_keys(to_jsonb(NEW)) AS field_name
78 WHERE to_jsonb(NEW) -> field_name IS DISTINCT FROM to_jsonb(OLD) -> field_name;
79 ELSE
80 changed := '[]'::jsonb;
81 END IF;
83 INSERT INTO object_streams_objectstreamevent (
84 subject_content_type_id,
85 subject_object_id,
86 source_content_type_id,
87 source_object_id,
88 source_history_content_type_id,
89 source_history_id,
90 facet,
91 op,
92 changed_fields,
93 before,
94 after,
95 metadata,
96 created_at
97 ) VALUES (
98 subject_type,
99 subject_row ->> '{subject_meta.pk.column}',
100 subject_type,
101 subject_row ->> '{subject_meta.pk.column}',
102 NULL,
103 '',
104 __FACET__,
105 subject_op,
106 changed,
107 NULL,
108 NULL,
109 jsonb_build_object('transaction_id', pg_current_xact_id()::text),
110 now()
111 ) RETURNING id INTO event_id;
113 PERFORM pg_notify(
114 coalesce(current_setting('__CHANNEL_SETTING__', true), __DEFAULT_CHANNEL__),
115 event_id::text
116 );
118 RETURN NULL;
119"""
122def _quote(value: str) -> str:
123 """Return a single-quoted SQL literal."""
124 escaped = value.replace("'", "''")
125 return f"'{escaped}'"
128class ObjectStreamTrigger(pgtrigger.Trigger):
129 """Write an outbox row whenever the model's table changes.
131 The row that changed is also the subject, which covers models whose own writes
132 are what subscribers care about. A source whose subject is a different object,
133 such as a workflow state row, needs its subject mapping expressed in SQL or left
134 to a Python producer.
136 ``changed_fields`` is computed by comparing the old and new row, so it reports
137 database column names. A foreign key appears as ``supplier_id`` rather than
138 ``supplier``, and it is populated even when the write did not pass
139 ``update_fields``, which the signal-based producers cannot do.
141 Every row carries the Postgres transaction id in ``metadata``, so a client can
142 tell that several events came from the same database transaction. It is not an
143 application action identifier because one action may span several transactions.
144 """
146 def __init__(self, *, facet: str = "object", channel: str | None = None, **kwargs):
147 self.facet = facet
148 self.channel = validate_notify_channel(channel) if channel is not None else None
149 kwargs.setdefault("when", pgtrigger.After)
150 kwargs.setdefault("operation", pgtrigger.Insert | pgtrigger.Update | pgtrigger.Delete)
151 kwargs.setdefault(
152 "declare",
153 [
154 ("subject_row", "JSONB"),
155 ("subject_type", "INTEGER"),
156 ("subject_op", "TEXT"),
157 ("changed", "JSONB"),
158 ("event_id", "BIGINT"),
159 ],
160 )
161 kwargs.setdefault("func", pgtrigger.Func(self._sql()))
162 super().__init__(**kwargs)
164 def _sql(self) -> str:
165 """Return the trigger body, leaving its model placeholders intact."""
166 return (
167 _FUNC.replace("__FACET__", _quote(self.facet))
168 .replace("__CHANNEL_SETTING__", NOTIFY_CHANNEL_SETTING)
169 .replace("__DEFAULT_CHANNEL__", _quote(self.channel or DEFAULT_NOTIFY_CHANNEL))
170 )
172 def get_func_template_kwargs(self, model):
173 """Render outbox references with the concrete model's identity."""
174 kwargs = super().get_func_template_kwargs(model)
175 subject_meta = model._meta.concrete_model._meta
176 if isinstance(subject_meta.pk, models.CompositePrimaryKey):
177 msg = f"ObjectStreamTrigger does not support composite primary keys ({subject_meta.label})."
178 raise ValueError(msg)
179 kwargs["subject_meta"] = subject_meta
180 return kwargs