Skip to content

Notifications Compatibility Internals

These pages document the legacy import paths that now forward to codex_django.messaging. Use Messaging internals for the canonical implementation docs.

Package facade

codex_django.notifications

Deprecated compatibility facade for :mod:codex_django.messaging.

Compatibility modules

codex_django.notifications.registry

Compatibility facade for :mod:codex_django.messaging.registry.

Classes

NotificationEventRegistry

Collect handlers that translate domain events into dispatch specs.

Source code in src/codex_django/messaging/registry.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class MessagingEventRegistry:
    """Collect handlers that translate domain events into dispatch specs."""

    def __init__(self) -> None:
        self._handlers: dict[str, list[object]] = {}

    def register(self, event_type: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
        """Register a handler for a logical notification event."""

        def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
            self._handlers.setdefault(event_type, []).append(fn)
            return fn

        return decorator

    def get_handlers(self, event_type: str) -> list[object]:
        """Return all handlers registered for one logical event type."""
        return list(self._handlers.get(event_type, ()))

    def build_specs(self, event_type: str, *args: Any, **kwargs: Any) -> list[NotificationDispatchSpec]:
        """Execute handlers and normalize their output into dispatch specs."""
        specs: list[NotificationDispatchSpec] = []
        for handler in self.get_handlers(event_type):
            result = cast(Callable[..., Any], handler)(*args, **kwargs)
            if result is None:
                continue
            if isinstance(result, NotificationDispatchSpec):
                specs.append(result)
                continue
            if isinstance(result, Iterable) and not isinstance(result, str | bytes | dict):
                for item in result:
                    if item is None:
                        continue
                    if not isinstance(item, NotificationDispatchSpec):
                        raise TypeError(
                            f"Notification handler for '{event_type}' returned non-spec item: {type(item)!r}"
                        )
                    specs.append(item)
                continue
            raise TypeError(
                f"Notification handler for '{event_type}' must return NotificationDispatchSpec, "
                f"an iterable of specs, or None; got {type(result)!r}"
            )
        return specs

    def email_template(
        self, event_type: str, *, channels: list[str]
    ) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
        """Register a template-mode builder."""

        def decorator(fn: Callable[..., dict[str, Any]]) -> Callable[..., NotificationDispatchSpec]:
            self._validate_registration(event_type, fn, mode="template")

            def builder(*args: Any, **kwargs: Any) -> NotificationDispatchSpec:
                data = self._validate_payload(event_type, fn(*args, **kwargs), mode="template")
                return NotificationDispatchSpec(
                    recipient_email=str(data["recipient_email"]),
                    subject_key=str(data["subject_key"]),
                    event_type=event_type,
                    channels=list(channels),
                    subject=str(data.get("subject", "")),
                    recipient_phone=data.get("recipient_phone"),
                    client_name=str(data.get("client_name", "")),
                    template_name=str(data["template_name"]),
                    language=str(data.get("language", "")),
                    mode="template",
                    context=dict(data.get("context", {})),
                )

            self.register(event_type)(builder)
            return builder

        return decorator

    def email_rendered(
        self, event_type: str, *, channels: list[str]
    ) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
        """Register a rendered-mode builder."""

        def decorator(fn: Callable[..., dict[str, Any]]) -> Callable[..., NotificationDispatchSpec]:
            self._validate_registration(event_type, fn, mode="rendered")

            def builder(*args: Any, **kwargs: Any) -> NotificationDispatchSpec:
                data = self._validate_payload(event_type, fn(*args, **kwargs), mode="rendered")
                return NotificationDispatchSpec(
                    recipient_email=str(data["recipient_email"]),
                    subject_key=str(data["subject_key"]),
                    event_type=event_type,
                    channels=list(channels),
                    subject=str(data.get("subject", "")),
                    recipient_phone=data.get("recipient_phone"),
                    client_name=str(data.get("client_name", "")),
                    language=str(data.get("language", "")),
                    mode="rendered",
                    html_content=str(data["html_content"]),
                    text_content=str(data.get("text_content", "")),
                    context=dict(data.get("context", {})),
                )

            self.register(event_type)(builder)
            return builder

        return decorator

    def _validate_registration(self, event_type: str, fn: Callable[..., dict[str, Any]], *, mode: str) -> None:
        try:
            sample = fn()
        except TypeError:
            return
        self._validate_payload(event_type, sample, mode=mode)

    def _validate_payload(self, event_type: str, data: dict[str, Any], *, mode: str) -> dict[str, Any]:
        if not isinstance(data, dict):
            raise TypeError(f"Handler for '{event_type}' must return dict, got {type(data)!r}")

        required = {"recipient_email", "subject_key"}
        if mode == "template":
            required.add("template_name")
            forbidden = {"html_content", "text_content", "mode"}
        else:
            required.add("html_content")
            forbidden = {"template_name", "mode"}

        missing = sorted(key for key in required if key not in data)
        if missing:
            raise TypeError(f"Handler for '{event_type}' is missing required keys: {', '.join(missing)}")

        present_forbidden = sorted(key for key in forbidden if key in data)
        if present_forbidden:
            raise TypeError(f"Handler for '{event_type}' returned forbidden keys: {', '.join(present_forbidden)}")

        return data
Methods:
register(event_type)

Register a handler for a logical notification event.

Source code in src/codex_django/messaging/registry.py 
17
18
19
20
21
22
23
24
def register(self, event_type: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Register a handler for a logical notification event."""

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        self._handlers.setdefault(event_type, []).append(fn)
        return fn

    return decorator
get_handlers(event_type)

Return all handlers registered for one logical event type.

Source code in src/codex_django/messaging/registry.py 
26
27
28
def get_handlers(self, event_type: str) -> list[object]:
    """Return all handlers registered for one logical event type."""
    return list(self._handlers.get(event_type, ()))
build_specs(event_type, *args, **kwargs)

Execute handlers and normalize their output into dispatch specs.

Source code in src/codex_django/messaging/registry.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
def build_specs(self, event_type: str, *args: Any, **kwargs: Any) -> list[NotificationDispatchSpec]:
    """Execute handlers and normalize their output into dispatch specs."""
    specs: list[NotificationDispatchSpec] = []
    for handler in self.get_handlers(event_type):
        result = cast(Callable[..., Any], handler)(*args, **kwargs)
        if result is None:
            continue
        if isinstance(result, NotificationDispatchSpec):
            specs.append(result)
            continue
        if isinstance(result, Iterable) and not isinstance(result, str | bytes | dict):
            for item in result:
                if item is None:
                    continue
                if not isinstance(item, NotificationDispatchSpec):
                    raise TypeError(
                        f"Notification handler for '{event_type}' returned non-spec item: {type(item)!r}"
                    )
                specs.append(item)
            continue
        raise TypeError(
            f"Notification handler for '{event_type}' must return NotificationDispatchSpec, "
            f"an iterable of specs, or None; got {type(result)!r}"
        )
    return specs
email_template(event_type, *, channels)

Register a template-mode builder.

Source code in src/codex_django/messaging/registry.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def email_template(
    self, event_type: str, *, channels: list[str]
) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
    """Register a template-mode builder."""

    def decorator(fn: Callable[..., dict[str, Any]]) -> Callable[..., NotificationDispatchSpec]:
        self._validate_registration(event_type, fn, mode="template")

        def builder(*args: Any, **kwargs: Any) -> NotificationDispatchSpec:
            data = self._validate_payload(event_type, fn(*args, **kwargs), mode="template")
            return NotificationDispatchSpec(
                recipient_email=str(data["recipient_email"]),
                subject_key=str(data["subject_key"]),
                event_type=event_type,
                channels=list(channels),
                subject=str(data.get("subject", "")),
                recipient_phone=data.get("recipient_phone"),
                client_name=str(data.get("client_name", "")),
                template_name=str(data["template_name"]),
                language=str(data.get("language", "")),
                mode="template",
                context=dict(data.get("context", {})),
            )

        self.register(event_type)(builder)
        return builder

    return decorator
email_rendered(event_type, *, channels)

Register a rendered-mode builder.

Source code in src/codex_django/messaging/registry.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def email_rendered(
    self, event_type: str, *, channels: list[str]
) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
    """Register a rendered-mode builder."""

    def decorator(fn: Callable[..., dict[str, Any]]) -> Callable[..., NotificationDispatchSpec]:
        self._validate_registration(event_type, fn, mode="rendered")

        def builder(*args: Any, **kwargs: Any) -> NotificationDispatchSpec:
            data = self._validate_payload(event_type, fn(*args, **kwargs), mode="rendered")
            return NotificationDispatchSpec(
                recipient_email=str(data["recipient_email"]),
                subject_key=str(data["subject_key"]),
                event_type=event_type,
                channels=list(channels),
                subject=str(data.get("subject", "")),
                recipient_phone=data.get("recipient_phone"),
                client_name=str(data.get("client_name", "")),
                language=str(data.get("language", "")),
                mode="rendered",
                html_content=str(data["html_content"]),
                text_content=str(data.get("text_content", "")),
                context=dict(data.get("context", {})),
            )

        self.register(event_type)(builder)
        return builder

    return decorator

Functions:

email_rendered(event_type, *, channels=None)

Register a rendered-mode event builder.

Source code in src/codex_django/messaging/registry.py 
160
161
162
163
164
def email_rendered(
    event_type: str, *, channels: list[str] | None = None
) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
    """Register a rendered-mode event builder."""
    return messaging_event_registry.email_rendered(event_type, channels=list(channels or ["email"]))

email_template(event_type, *, channels=None)

Register a template-mode event builder.

Source code in src/codex_django/messaging/registry.py 
153
154
155
156
157
def email_template(
    event_type: str, *, channels: list[str] | None = None
) -> Callable[[Callable[..., dict[str, Any]]], Callable[..., NotificationDispatchSpec]]:
    """Register a template-mode event builder."""
    return messaging_event_registry.email_template(event_type, channels=list(channels or ["email"]))

notification_handler(event_type)

Convenience decorator backed by the global messaging event registry.

Source code in src/codex_django/messaging/registry.py 
148
149
150
def notification_handler(event_type: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Convenience decorator backed by the global messaging event registry."""
    return messaging_event_registry.register(event_type)

codex_django.notifications.service

Compatibility facade for :mod:codex_django.messaging.service.

Classes

BaseNotificationEngine

Coordinate subject lookup, payload building, and queue dispatch.

Notes

task_name identifies the worker task that performs delivery. mode controls whether the worker renders a template or receives pre-rendered content directly.

Source code in src/codex_django/messaging/service.py 
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
class BaseMessagingEngine:
    """Coordinate subject lookup, payload building, and queue dispatch.

    Notes:
        ``task_name`` identifies the worker task that performs delivery.
        ``mode`` controls whether the worker renders a template or receives
        pre-rendered content directly.
    """

    task_name: str = "send_universal_notification_task"
    mode: str = "template"  # default mode for all dispatch() calls
    default_language: str = "de"

    def __init__(
        self,
        queue_adapter: QueueAdapterProtocol,
        cache_adapter: Any,
        i18n_adapter: Any,
        selector: Any,
    ) -> None:
        """Initialize the engine with project-provided infrastructure adapters.

        Args:
            queue_adapter: Adapter responsible for queueing or delivering work.
            cache_adapter: Cache adapter used by the project selector layer.
            i18n_adapter: Adapter that manages temporary language overrides.
            selector: Content selector used to resolve localized subjects.
        """
        self._queue = queue_adapter
        self._cache = cache_adapter
        self._i18n = i18n_adapter
        self._selector = selector
        self._builder = _get_builder()

    # ------------------------------------------------------------------
    # Main entry point
    # ------------------------------------------------------------------

    def dispatch(
        self,
        *,
        recipient_email: str,
        recipient_phone: str | None = None,
        client_name: str = "",
        template_name: str = "",
        event_type: str,
        channels: list[str],
        language: str = "",
        subject_key: str,
        subject: str = "",
        mode: str | None = None,
        # Mode 2 only:
        html_content: str = "",
        text_content: str = "",
        **context: Any,
    ) -> str | None:
        """Resolve content metadata, build a payload, and enqueue a message.

        Args:
            recipient_email: Destination email address.
            recipient_phone: Optional phone number for SMS/WhatsApp channels.
            client_name: Optional human-readable recipient name.
            template_name: Worker-side template path for ``template`` mode.
            event_type: Logical messaging event identifier.
            channels: Delivery channels to request from the worker.
            language: Language code used for subject lookup and payload metadata.
                Falls back to ``default_language`` when empty.
            subject_key: Content key used to resolve the localized subject.
            subject: Optional explicit subject override. When provided, the
                selector lookup is skipped.
            mode: Optional per-call override for ``template`` or ``rendered`` mode.
            html_content: Pre-rendered HTML body for ``rendered`` mode.
            text_content: Optional plain-text fallback for ``rendered`` mode.
            **context: Extra payload context passed to the worker.

        Returns:
            Queue job ID when the adapter returns one, otherwise ``None``.
        """
        effective_mode = mode or self.mode
        lang = language or self.default_language

        resolved_subject = subject or self._selector.get(subject_key, lang) or ""
        notification_id = str(uuid.uuid4())

        if effective_mode == "template":
            payload = self._builder.build_template(
                notification_id=notification_id,
                recipient_email=recipient_email,
                recipient_phone=recipient_phone,
                client_name=client_name,
                template_name=template_name,
                subject=resolved_subject,
                event_type=event_type,
                context_data=context,
                channels=channels,
                language=lang,
            )
        else:
            payload = self._builder.build_rendered(
                notification_id=notification_id,
                recipient_email=recipient_email,
                recipient_phone=recipient_phone,
                client_name=client_name,
                html_content=html_content,
                text_content=text_content,
                subject=resolved_subject,
                event_type=event_type,
                channels=channels,
                language=lang,
                context_data=context,
            )

        return self._queue.enqueue(self.task_name, payload)

    def dispatch_spec(self, spec: NotificationDispatchSpec) -> str | None:
        """Dispatch one prebuilt spec produced by a domain handler."""
        return self.dispatch(
            recipient_email=spec.recipient_email,
            recipient_phone=spec.recipient_phone,
            client_name=spec.client_name,
            template_name=spec.template_name,
            event_type=spec.event_type,
            channels=spec.channels,
            language=spec.language,
            subject_key=spec.subject_key,
            subject=spec.subject,
            mode=spec.mode,
            html_content=spec.html_content,
            text_content=spec.text_content,
            **spec.context,
        )

    def dispatch_event(self, event_type: str, *args: Any, **kwargs: Any) -> list[str | None]:
        """Resolve registered handlers for an event and dispatch every resulting spec."""
        specs = messaging_event_registry.build_specs(event_type, *args, **kwargs)
        return [self.dispatch_spec(spec) for spec in specs]

    async def adispatch(
        self,
        *,
        recipient_email: str,
        recipient_phone: str | None = None,
        client_name: str = "",
        template_name: str = "",
        event_type: str,
        channels: list[str],
        language: str = "",
        subject_key: str,
        subject: str = "",
        mode: str | None = None,
        html_content: str = "",
        text_content: str = "",
        **context: Any,
    ) -> str | None:
        """Asynchronous counterpart to :meth:`dispatch` for async call sites.

        Args:
            recipient_email: Destination email address.
            recipient_phone: Optional phone number for SMS/WhatsApp channels.
            client_name: Optional human-readable recipient name.
            template_name: Worker-side template path for ``template`` mode.
            event_type: Logical messaging event identifier.
            channels: Delivery channels to request from the worker.
            language: Language code used for subject lookup and payload metadata.
                Falls back to ``default_language`` when empty.
            subject_key: Content key used to resolve the localized subject.
            subject: Optional explicit subject override. When provided, the
                selector lookup is skipped.
            mode: Optional per-call override for ``template`` or ``rendered`` mode.
            html_content: Pre-rendered HTML body for ``rendered`` mode.
            text_content: Optional plain-text fallback for ``rendered`` mode.
            **context: Extra payload context passed to the worker.

        Returns:
            Queue job ID when the adapter returns one, otherwise ``None``.
        """
        effective_mode = mode or self.mode
        lang = language or self.default_language

        resolved_subject = subject or self._selector.get(subject_key, lang) or ""
        notification_id = str(uuid.uuid4())

        if effective_mode == "template":
            payload = self._builder.build_template(
                notification_id=notification_id,
                recipient_email=recipient_email,
                recipient_phone=recipient_phone,
                client_name=client_name,
                template_name=template_name,
                subject=resolved_subject,
                event_type=event_type,
                context_data=context,
                channels=channels,
                language=lang,
            )
        else:
            payload = self._builder.build_rendered(
                notification_id=notification_id,
                recipient_email=recipient_email,
                recipient_phone=recipient_phone,
                client_name=client_name,
                html_content=html_content,
                text_content=text_content,
                subject=resolved_subject,
                event_type=event_type,
                channels=channels,
                language=lang,
                context_data=context,
            )

        return await self._queue.aenqueue(self.task_name, payload)

    async def adispatch_spec(self, spec: NotificationDispatchSpec) -> str | None:
        """Async counterpart to :meth:`dispatch_spec`."""
        return await self.adispatch(
            recipient_email=spec.recipient_email,
            recipient_phone=spec.recipient_phone,
            client_name=spec.client_name,
            template_name=spec.template_name,
            event_type=spec.event_type,
            channels=spec.channels,
            language=spec.language,
            subject_key=spec.subject_key,
            subject=spec.subject,
            mode=spec.mode,
            html_content=spec.html_content,
            text_content=spec.text_content,
            **spec.context,
        )

    async def adispatch_event(self, event_type: str, *args: Any, **kwargs: Any) -> list[str | None]:
        """Async counterpart to :meth:`dispatch_event`."""
        specs = messaging_event_registry.build_specs(event_type, *args, **kwargs)
        results: list[str | None] = []
        for spec in specs:
            results.append(await self.adispatch_spec(spec))
        return results
Methods:
__init__(queue_adapter, cache_adapter, i18n_adapter, selector)

Initialize the engine with project-provided infrastructure adapters.

Parameters:

Name Type Description Default
queue_adapter QueueAdapterProtocol

Adapter responsible for queueing or delivering work.

 required
cache_adapter Any

Cache adapter used by the project selector layer.

 required
i18n_adapter Any

Adapter that manages temporary language overrides.

 required
selector Any

Content selector used to resolve localized subjects.

 required
Source code in src/codex_django/messaging/service.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def __init__(
    self,
    queue_adapter: QueueAdapterProtocol,
    cache_adapter: Any,
    i18n_adapter: Any,
    selector: Any,
) -> None:
    """Initialize the engine with project-provided infrastructure adapters.

    Args:
        queue_adapter: Adapter responsible for queueing or delivering work.
        cache_adapter: Cache adapter used by the project selector layer.
        i18n_adapter: Adapter that manages temporary language overrides.
        selector: Content selector used to resolve localized subjects.
    """
    self._queue = queue_adapter
    self._cache = cache_adapter
    self._i18n = i18n_adapter
    self._selector = selector
    self._builder = _get_builder()
dispatch(*, recipient_email, recipient_phone=None, client_name='', template_name='', event_type, channels, language='', subject_key, subject='', mode=None, html_content='', text_content='', **context)

Resolve content metadata, build a payload, and enqueue a message.

Parameters:

Name Type Description Default
recipient_email str

Destination email address.

 required
recipient_phone str | None

Optional phone number for SMS/WhatsApp channels.

 None
client_name str

Optional human-readable recipient name.

 ''
template_name str

Worker-side template path for template mode.

 ''
event_type str

Logical messaging event identifier.

 required
channels list[str]

Delivery channels to request from the worker.

 required
language str

Language code used for subject lookup and payload metadata. Falls back to default_language when empty.

 ''
subject_key str

Content key used to resolve the localized subject.

 required
subject str

Optional explicit subject override. When provided, the selector lookup is skipped.

 ''
mode str | None

Optional per-call override for template or rendered mode.

 None
html_content str

Pre-rendered HTML body for rendered mode.

 ''
text_content str

Optional plain-text fallback for rendered mode.

 ''
**context Any

Extra payload context passed to the worker.

 {}

Returns:

Type Description
str | None

Queue job ID when the adapter returns one, otherwise None.
Source code in src/codex_django/messaging/service.py 
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def dispatch(
    self,
    *,
    recipient_email: str,
    recipient_phone: str | None = None,
    client_name: str = "",
    template_name: str = "",
    event_type: str,
    channels: list[str],
    language: str = "",
    subject_key: str,
    subject: str = "",
    mode: str | None = None,
    # Mode 2 only:
    html_content: str = "",
    text_content: str = "",
    **context: Any,
) -> str | None:
    """Resolve content metadata, build a payload, and enqueue a message.

    Args:
        recipient_email: Destination email address.
        recipient_phone: Optional phone number for SMS/WhatsApp channels.
        client_name: Optional human-readable recipient name.
        template_name: Worker-side template path for ``template`` mode.
        event_type: Logical messaging event identifier.
        channels: Delivery channels to request from the worker.
        language: Language code used for subject lookup and payload metadata.
            Falls back to ``default_language`` when empty.
        subject_key: Content key used to resolve the localized subject.
        subject: Optional explicit subject override. When provided, the
            selector lookup is skipped.
        mode: Optional per-call override for ``template`` or ``rendered`` mode.
        html_content: Pre-rendered HTML body for ``rendered`` mode.
        text_content: Optional plain-text fallback for ``rendered`` mode.
        **context: Extra payload context passed to the worker.

    Returns:
        Queue job ID when the adapter returns one, otherwise ``None``.
    """
    effective_mode = mode or self.mode
    lang = language or self.default_language

    resolved_subject = subject or self._selector.get(subject_key, lang) or ""
    notification_id = str(uuid.uuid4())

    if effective_mode == "template":
        payload = self._builder.build_template(
            notification_id=notification_id,
            recipient_email=recipient_email,
            recipient_phone=recipient_phone,
            client_name=client_name,
            template_name=template_name,
            subject=resolved_subject,
            event_type=event_type,
            context_data=context,
            channels=channels,
            language=lang,
        )
    else:
        payload = self._builder.build_rendered(
            notification_id=notification_id,
            recipient_email=recipient_email,
            recipient_phone=recipient_phone,
            client_name=client_name,
            html_content=html_content,
            text_content=text_content,
            subject=resolved_subject,
            event_type=event_type,
            channels=channels,
            language=lang,
            context_data=context,
        )

    return self._queue.enqueue(self.task_name, payload)
dispatch_spec(spec)

Dispatch one prebuilt spec produced by a domain handler.

Source code in src/codex_django/messaging/service.py 
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def dispatch_spec(self, spec: NotificationDispatchSpec) -> str | None:
    """Dispatch one prebuilt spec produced by a domain handler."""
    return self.dispatch(
        recipient_email=spec.recipient_email,
        recipient_phone=spec.recipient_phone,
        client_name=spec.client_name,
        template_name=spec.template_name,
        event_type=spec.event_type,
        channels=spec.channels,
        language=spec.language,
        subject_key=spec.subject_key,
        subject=spec.subject,
        mode=spec.mode,
        html_content=spec.html_content,
        text_content=spec.text_content,
        **spec.context,
    )
dispatch_event(event_type, *args, **kwargs)

Resolve registered handlers for an event and dispatch every resulting spec.

Source code in src/codex_django/messaging/service.py 
174
175
176
177
def dispatch_event(self, event_type: str, *args: Any, **kwargs: Any) -> list[str | None]:
    """Resolve registered handlers for an event and dispatch every resulting spec."""
    specs = messaging_event_registry.build_specs(event_type, *args, **kwargs)
    return [self.dispatch_spec(spec) for spec in specs]
adispatch(*, recipient_email, recipient_phone=None, client_name='', template_name='', event_type, channels, language='', subject_key, subject='', mode=None, html_content='', text_content='', **context) async

Asynchronous counterpart to :meth:dispatch for async call sites.

Parameters:

Name Type Description Default
recipient_email str

Destination email address.

 required
recipient_phone str | None

Optional phone number for SMS/WhatsApp channels.

 None
client_name str

Optional human-readable recipient name.

 ''
template_name str

Worker-side template path for template mode.

 ''
event_type str

Logical messaging event identifier.

 required
channels list[str]

Delivery channels to request from the worker.

 required
language str

Language code used for subject lookup and payload metadata. Falls back to default_language when empty.

 ''
subject_key str

Content key used to resolve the localized subject.

 required
subject str

Optional explicit subject override. When provided, the selector lookup is skipped.

 ''
mode str | None

Optional per-call override for template or rendered mode.

 None
html_content str

Pre-rendered HTML body for rendered mode.

 ''
text_content str

Optional plain-text fallback for rendered mode.

 ''
**context Any

Extra payload context passed to the worker.

 {}

Returns:

Type Description
str | None

Queue job ID when the adapter returns one, otherwise None.
Source code in src/codex_django/messaging/service.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
async def adispatch(
    self,
    *,
    recipient_email: str,
    recipient_phone: str | None = None,
    client_name: str = "",
    template_name: str = "",
    event_type: str,
    channels: list[str],
    language: str = "",
    subject_key: str,
    subject: str = "",
    mode: str | None = None,
    html_content: str = "",
    text_content: str = "",
    **context: Any,
) -> str | None:
    """Asynchronous counterpart to :meth:`dispatch` for async call sites.

    Args:
        recipient_email: Destination email address.
        recipient_phone: Optional phone number for SMS/WhatsApp channels.
        client_name: Optional human-readable recipient name.
        template_name: Worker-side template path for ``template`` mode.
        event_type: Logical messaging event identifier.
        channels: Delivery channels to request from the worker.
        language: Language code used for subject lookup and payload metadata.
            Falls back to ``default_language`` when empty.
        subject_key: Content key used to resolve the localized subject.
        subject: Optional explicit subject override. When provided, the
            selector lookup is skipped.
        mode: Optional per-call override for ``template`` or ``rendered`` mode.
        html_content: Pre-rendered HTML body for ``rendered`` mode.
        text_content: Optional plain-text fallback for ``rendered`` mode.
        **context: Extra payload context passed to the worker.

    Returns:
        Queue job ID when the adapter returns one, otherwise ``None``.
    """
    effective_mode = mode or self.mode
    lang = language or self.default_language

    resolved_subject = subject or self._selector.get(subject_key, lang) or ""
    notification_id = str(uuid.uuid4())

    if effective_mode == "template":
        payload = self._builder.build_template(
            notification_id=notification_id,
            recipient_email=recipient_email,
            recipient_phone=recipient_phone,
            client_name=client_name,
            template_name=template_name,
            subject=resolved_subject,
            event_type=event_type,
            context_data=context,
            channels=channels,
            language=lang,
        )
    else:
        payload = self._builder.build_rendered(
            notification_id=notification_id,
            recipient_email=recipient_email,
            recipient_phone=recipient_phone,
            client_name=client_name,
            html_content=html_content,
            text_content=text_content,
            subject=resolved_subject,
            event_type=event_type,
            channels=channels,
            language=lang,
            context_data=context,
        )

    return await self._queue.aenqueue(self.task_name, payload)
adispatch_spec(spec) async

Async counterpart to :meth:dispatch_spec.

Source code in src/codex_django/messaging/service.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
async def adispatch_spec(self, spec: NotificationDispatchSpec) -> str | None:
    """Async counterpart to :meth:`dispatch_spec`."""
    return await self.adispatch(
        recipient_email=spec.recipient_email,
        recipient_phone=spec.recipient_phone,
        client_name=spec.client_name,
        template_name=spec.template_name,
        event_type=spec.event_type,
        channels=spec.channels,
        language=spec.language,
        subject_key=spec.subject_key,
        subject=spec.subject,
        mode=spec.mode,
        html_content=spec.html_content,
        text_content=spec.text_content,
        **spec.context,
    )
adispatch_event(event_type, *args, **kwargs) async

Async counterpart to :meth:dispatch_event.

Source code in src/codex_django/messaging/service.py 
272
273
274
275
276
277
278
async def adispatch_event(self, event_type: str, *args: Any, **kwargs: Any) -> list[str | None]:
    """Async counterpart to :meth:`dispatch_event`."""
    specs = messaging_event_registry.build_specs(event_type, *args, **kwargs)
    results: list[str | None] = []
    for spec in specs:
        results.append(await self.adispatch_spec(spec))
    return results

codex_django.notifications.builder

Compatibility facade for :mod:codex_django.messaging.builder.

Classes

NotificationPayloadBuilder

Build serializable payload dictionaries for messaging queue tasks.

The builder is framework-agnostic and intentionally avoids Django imports so it can be reused in synchronous, asynchronous, and worker contexts.

Source code in src/codex_django/messaging/builder.py 
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class MessagingPayloadBuilder:
    """Build serializable payload dictionaries for messaging queue tasks.

    The builder is framework-agnostic and intentionally avoids Django imports
    so it can be reused in synchronous, asynchronous, and worker contexts.
    """

    def build_template(
        self,
        *,
        notification_id: str,
        recipient_email: str,
        recipient_phone: str | None = None,
        client_name: str = "",
        template_name: str,
        subject: str,
        event_type: str,
        context_data: dict[str, Any],
        channels: list[str],
        language: str = "de",
    ) -> dict[str, Any]:
        """Build a payload where the worker renders the template itself.

        Args:
            notification_id: Unique notification identifier.
            recipient_email: Destination email address.
            recipient_phone: Optional phone number for SMS/WhatsApp channels.
            client_name: Optional human-readable recipient name.
            template_name: Worker-side template path to render.
            subject: Localized notification subject.
            event_type: Logical event identifier.
            context_data: Extra template context data.
            channels: Delivery channels requested for the notification.
            language: Language code attached to the payload.

        Returns:
            A plain dictionary compatible with the template-based notification
            DTO expected by the queue worker.
        """
        return {
            "mode": "template",
            "notification_id": notification_id,
            "recipient_email": recipient_email,
            "recipient_phone": recipient_phone,
            "client_name": client_name,
            "template_name": template_name,
            "subject": subject,
            "event_type": event_type,
            "context_data": context_data,
            "channels": channels,
            "language": language,
        }

    def build_rendered(
        self,
        *,
        notification_id: str,
        recipient_email: str,
        recipient_phone: str | None = None,
        client_name: str = "",
        html_content: str,
        text_content: str = "",
        subject: str,
        event_type: str,
        channels: list[str],
        language: str = "de",
        context_data: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Build a payload that already contains rendered notification content.

        Args:
            notification_id: Unique notification identifier.
            recipient_email: Destination email address.
            recipient_phone: Optional phone number for SMS/WhatsApp channels.
            client_name: Optional human-readable recipient name.
            html_content: Pre-rendered HTML body.
            text_content: Optional plain-text body.
            subject: Localized notification subject.
            event_type: Logical event identifier.
            channels: Delivery channels requested for the notification.
            language: Language code attached to the payload.

        Returns:
            A plain dictionary compatible with the rendered notification DTO
            expected by the queue worker.
        """
        return {
            "mode": "rendered",
            "notification_id": notification_id,
            "recipient_email": recipient_email,
            "recipient_phone": recipient_phone,
            "client_name": client_name,
            "html_content": html_content,
            "text_content": text_content,
            "subject": subject,
            "event_type": event_type,
            "channels": channels,
            "language": language,
            "context_data": context_data or {},
        }
Methods:
build_template(*, notification_id, recipient_email, recipient_phone=None, client_name='', template_name, subject, event_type, context_data, channels, language='de')

Build a payload where the worker renders the template itself.

Parameters:

Name Type Description Default
notification_id str

Unique notification identifier.

 required
recipient_email str

Destination email address.

 required
recipient_phone str | None

Optional phone number for SMS/WhatsApp channels.

 None
client_name str

Optional human-readable recipient name.

 ''
template_name str

Worker-side template path to render.

 required
subject str

Localized notification subject.

 required
event_type str

Logical event identifier.

 required
context_data dict[str, Any]

Extra template context data.

 required
channels list[str]

Delivery channels requested for the notification.

 required
language str

Language code attached to the payload.

 'de'

Returns:

Type Description
dict[str, Any]

A plain dictionary compatible with the template-based notification
dict[str, Any]

DTO expected by the queue worker.
Source code in src/codex_django/messaging/builder.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def build_template(
    self,
    *,
    notification_id: str,
    recipient_email: str,
    recipient_phone: str | None = None,
    client_name: str = "",
    template_name: str,
    subject: str,
    event_type: str,
    context_data: dict[str, Any],
    channels: list[str],
    language: str = "de",
) -> dict[str, Any]:
    """Build a payload where the worker renders the template itself.

    Args:
        notification_id: Unique notification identifier.
        recipient_email: Destination email address.
        recipient_phone: Optional phone number for SMS/WhatsApp channels.
        client_name: Optional human-readable recipient name.
        template_name: Worker-side template path to render.
        subject: Localized notification subject.
        event_type: Logical event identifier.
        context_data: Extra template context data.
        channels: Delivery channels requested for the notification.
        language: Language code attached to the payload.

    Returns:
        A plain dictionary compatible with the template-based notification
        DTO expected by the queue worker.
    """
    return {
        "mode": "template",
        "notification_id": notification_id,
        "recipient_email": recipient_email,
        "recipient_phone": recipient_phone,
        "client_name": client_name,
        "template_name": template_name,
        "subject": subject,
        "event_type": event_type,
        "context_data": context_data,
        "channels": channels,
        "language": language,
    }
build_rendered(*, notification_id, recipient_email, recipient_phone=None, client_name='', html_content, text_content='', subject, event_type, channels, language='de', context_data=None)

Build a payload that already contains rendered notification content.

Parameters:

Name Type Description Default
notification_id str

Unique notification identifier.

 required
recipient_email str

Destination email address.

 required
recipient_phone str | None

Optional phone number for SMS/WhatsApp channels.

 None
client_name str

Optional human-readable recipient name.

 ''
html_content str

Pre-rendered HTML body.

 required
text_content str

Optional plain-text body.

 ''
subject str

Localized notification subject.

 required
event_type str

Logical event identifier.

 required
channels list[str]

Delivery channels requested for the notification.

 required
language str

Language code attached to the payload.

 'de'

Returns:

Type Description
dict[str, Any]

A plain dictionary compatible with the rendered notification DTO
dict[str, Any]

expected by the queue worker.
Source code in src/codex_django/messaging/builder.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def build_rendered(
    self,
    *,
    notification_id: str,
    recipient_email: str,
    recipient_phone: str | None = None,
    client_name: str = "",
    html_content: str,
    text_content: str = "",
    subject: str,
    event_type: str,
    channels: list[str],
    language: str = "de",
    context_data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Build a payload that already contains rendered notification content.

    Args:
        notification_id: Unique notification identifier.
        recipient_email: Destination email address.
        recipient_phone: Optional phone number for SMS/WhatsApp channels.
        client_name: Optional human-readable recipient name.
        html_content: Pre-rendered HTML body.
        text_content: Optional plain-text body.
        subject: Localized notification subject.
        event_type: Logical event identifier.
        channels: Delivery channels requested for the notification.
        language: Language code attached to the payload.

    Returns:
        A plain dictionary compatible with the rendered notification DTO
        expected by the queue worker.
    """
    return {
        "mode": "rendered",
        "notification_id": notification_id,
        "recipient_email": recipient_email,
        "recipient_phone": recipient_phone,
        "client_name": client_name,
        "html_content": html_content,
        "text_content": text_content,
        "subject": subject,
        "event_type": event_type,
        "channels": channels,
        "language": language,
        "context_data": context_data or {},
    }

codex_django.notifications.selector

Compatibility facade for :mod:codex_django.messaging.selector.

Classes

BaseEmailContentSelector

Retrieve messaging text blocks with cache and language awareness.

The selector is model-agnostic: the concrete content model, cache adapter, and i18n adapter are injected by the project at construction time.

Notes

Default cache TTL is one hour.

Source code in src/codex_django/messaging/selector.py 
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
class BaseEmailContentSelector:
    """Retrieve messaging text blocks with cache and language awareness.

    The selector is model-agnostic: the concrete content model, cache adapter,
    and i18n adapter are injected by the project at construction time.

    Notes:
        Default cache TTL is one hour.
    """

    cache_timeout: int = 3600
    cache_key_prefix: str = ""
    default_language: str = "de"

    def __init__(
        self,
        model: Any,
        cache_adapter: Any,
        i18n_adapter: Any,
    ) -> None:
        """Initialize the selector with project-specific dependencies.

        Args:
            model: Concrete content model class used for lookups.
            cache_adapter: Cache adapter implementing ``get`` and ``set``.
            i18n_adapter: Adapter exposing a ``translation_override`` context manager.
        """
        self._model = model
        self._cache = cache_adapter
        self._i18n = i18n_adapter

    def get(self, key: str, language: str = "") -> str | None:
        """Return the messaging text for a key in the requested language.

        Args:
            key: Logical content key stored in the backing model.
            language: Target Django language code. Falls back to
                ``default_language`` when empty.

        Returns:
            The resolved text value, or ``None`` when no matching content block
            exists.
        """
        lang = language or self.default_language
        cache_key = self._cache_key(key, lang)

        cached = self._cache.get(cache_key)
        if cached is not None:
            return cast(str | None, cached)

        value = self._fetch_from_db(key, lang)
        if value is not None:
            self._cache.set(cache_key, value, self.cache_timeout)

        return value

    def invalidate(self, key: str, language: str = "") -> None:
        """Invalidate a single cached messaging content entry.

        Args:
            key: Logical content key stored in the backing model.
            language: Language code associated with the cached value. Falls
                back to ``default_language`` when empty.
        """
        lang = language or self.default_language
        self._cache.set(self._cache_key(key, lang), "", timeout=0)

    # ------------------------------------------------------------------
    # Internal
    # ------------------------------------------------------------------

    def _cache_key(self, key: str, language: str) -> str:
        """Build the cache key used for one localized content entry."""
        return f"{self.cache_key_prefix}{key}:{language}"

    def _fetch_from_db(self, key: str, language: str) -> str | None:
        """Fetch a localized content value directly from the database."""
        try:
            with self._i18n.translation_override(language):
                obj = self._model.objects.get(key=key)
                return cast(str, obj.text)
        except self._model.DoesNotExist:
            log.warning("EmailContent key=%r language=%r not found in DB", key, language)
            return None
Methods:
__init__(model, cache_adapter, i18n_adapter)

Initialize the selector with project-specific dependencies.

Parameters:

Name Type Description Default
model Any

Concrete content model class used for lookups.

 required
cache_adapter Any

Cache adapter implementing get and set.

 required
i18n_adapter Any

Adapter exposing a translation_override context manager.

 required
Source code in src/codex_django/messaging/selector.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
def __init__(
    self,
    model: Any,
    cache_adapter: Any,
    i18n_adapter: Any,
) -> None:
    """Initialize the selector with project-specific dependencies.

    Args:
        model: Concrete content model class used for lookups.
        cache_adapter: Cache adapter implementing ``get`` and ``set``.
        i18n_adapter: Adapter exposing a ``translation_override`` context manager.
    """
    self._model = model
    self._cache = cache_adapter
    self._i18n = i18n_adapter
get(key, language='')

Return the messaging text for a key in the requested language.

Parameters:

Name Type Description Default
key str

Logical content key stored in the backing model.

 required
language str

Target Django language code. Falls back to default_language when empty.

 ''

Returns:

Type Description
str | None

The resolved text value, or None when no matching content block
str | None

exists.
Source code in src/codex_django/messaging/selector.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def get(self, key: str, language: str = "") -> str | None:
    """Return the messaging text for a key in the requested language.

    Args:
        key: Logical content key stored in the backing model.
        language: Target Django language code. Falls back to
            ``default_language`` when empty.

    Returns:
        The resolved text value, or ``None`` when no matching content block
        exists.
    """
    lang = language or self.default_language
    cache_key = self._cache_key(key, lang)

    cached = self._cache.get(cache_key)
    if cached is not None:
        return cast(str | None, cached)

    value = self._fetch_from_db(key, lang)
    if value is not None:
        self._cache.set(cache_key, value, self.cache_timeout)

    return value
invalidate(key, language='')

Invalidate a single cached messaging content entry.

Parameters:

Name Type Description Default
key str

Logical content key stored in the backing model.

 required
language str

Language code associated with the cached value. Falls back to default_language when empty.

 ''
Source code in src/codex_django/messaging/selector.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def invalidate(self, key: str, language: str = "") -> None:
    """Invalidate a single cached messaging content entry.

    Args:
        key: Logical content key stored in the backing model.
        language: Language code associated with the cached value. Falls
            back to ``default_language`` when empty.
    """
    lang = language or self.default_language
    self._cache.set(self._cache_key(key, lang), "", timeout=0)

codex_django.notifications.contracts

Compatibility facade for :mod:codex_django.messaging.contracts.

Classes

ContentSelectorProtocol

Bases: Protocol

Localized content lookup contract used by messaging dispatch.

Source code in src/codex_django/messaging/contracts.py 
37
38
39
40
class ContentSelectorProtocol(Protocol):
    """Localized content lookup contract used by messaging dispatch."""

    def get(self, key: str, language: str = "") -> str | None: ...

NotificationDispatchSpec dataclass

Normalized dispatch contract for one message delivery request.

Source code in src/codex_django/messaging/contracts.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
@dataclass(frozen=True)
class NotificationDispatchSpec:
    """Normalized dispatch contract for one message delivery request."""

    recipient_email: str
    subject_key: str
    event_type: str
    channels: list[str]
    subject: str = ""
    recipient_phone: str | None = None
    client_name: str = ""
    template_name: str = ""
    language: str = ""
    mode: str | None = None
    html_content: str = ""
    text_content: str = ""
    context: dict[str, Any] = field(default_factory=dict)

NotificationEventHandler

Bases: Protocol

Decorator-friendly handler that yields one or more dispatch specs.

Source code in src/codex_django/messaging/contracts.py 
43
44
45
46
47
48
class NotificationEventHandler(Protocol):
    """Decorator-friendly handler that yields one or more dispatch specs."""

    def __call__(
        self, *args: Any, **kwargs: Any
    ) -> NotificationDispatchSpec | Iterable[NotificationDispatchSpec] | None: ...

QueueAdapterProtocol

Bases: Protocol

Minimal queue adapter contract required by the messaging engine.

Source code in src/codex_django/messaging/contracts.py 
29
30
31
32
33
34
class QueueAdapterProtocol(Protocol):
    """Minimal queue adapter contract required by the messaging engine."""

    def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None: ...

    async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None: ...