Skip to content

Messaging Internals

codex_django.messaging

codex_django.messaging

Django-specific messaging infrastructure for Codex projects.

Classes

DjangoArqClient

Thin Django-friendly wrapper around codex-platform's ARQ adapter.

Source code in src/codex_django/messaging/adapters/arq_client.py 
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
class DjangoArqClient:
    """Thin Django-friendly wrapper around codex-platform's ARQ adapter."""

    def __init__(
        self,
        *,
        adapter: _ArqAdapterProtocol | None = None,
        pool: Any | None = None,
        redis_settings: Any | None = None,
    ) -> None:
        self._pool: Any | None = pool
        self._adapter: _ArqAdapterProtocol | None = adapter
        self._redis_settings: Any | None = redis_settings

    @staticmethod
    def build_redis_settings_from_django() -> Any:
        """Build ARQ RedisSettings from Django settings as a convenience fallback."""
        from arq.connections import RedisSettings

        redis_url = getattr(settings, "ARQ_REDIS_URL", None) or getattr(settings, "REDIS_URL", None)
        if redis_url:
            parsed = urlparse(str(redis_url))
            database = int(parsed.path.lstrip("/") or "0")
            return RedisSettings(
                host=parsed.hostname or "localhost",
                port=parsed.port or 6379,
                username=parsed.username,
                password=parsed.password,
                database=database,
                ssl=parsed.scheme == "rediss",
            )

        return RedisSettings(
            host=getattr(settings, "REDIS_HOST", "localhost"),
            port=int(getattr(settings, "REDIS_PORT", 6379)),
            password=getattr(settings, "REDIS_PASSWORD", None),
            database=int(getattr(settings, "REDIS_DB", 0)),
        )

    async def _get_adapter(self) -> _ArqAdapterProtocol:
        """Lazily create the underlying codex-platform ARQ adapter."""
        if self._adapter is None:
            from arq.connections import create_pool
            from codex_platform.notifications.delivery import ArqNotificationAdapter

            if self._pool is None:
                redis_settings = self._redis_settings or self.build_redis_settings_from_django()
                self._pool = await create_pool(redis_settings)
            self._adapter = cast(_ArqAdapterProtocol, ArqNotificationAdapter(self._pool))
        return self._adapter

    def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
        """Enqueue a task synchronously via the platform ARQ adapter."""
        adapter = async_to_sync(self._get_adapter)()
        return adapter.enqueue(task_name, payload)

    async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
        """Enqueue a task asynchronously via the platform ARQ adapter."""
        adapter = await self._get_adapter()
        return await adapter.enqueue_async(task_name, payload)
Methods:
build_redis_settings_from_django() staticmethod

Build ARQ RedisSettings from Django settings as a convenience fallback.

Source code in src/codex_django/messaging/adapters/arq_client.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@staticmethod
def build_redis_settings_from_django() -> Any:
    """Build ARQ RedisSettings from Django settings as a convenience fallback."""
    from arq.connections import RedisSettings

    redis_url = getattr(settings, "ARQ_REDIS_URL", None) or getattr(settings, "REDIS_URL", None)
    if redis_url:
        parsed = urlparse(str(redis_url))
        database = int(parsed.path.lstrip("/") or "0")
        return RedisSettings(
            host=parsed.hostname or "localhost",
            port=parsed.port or 6379,
            username=parsed.username,
            password=parsed.password,
            database=database,
            ssl=parsed.scheme == "rediss",
        )

    return RedisSettings(
        host=getattr(settings, "REDIS_HOST", "localhost"),
        port=int(getattr(settings, "REDIS_PORT", 6379)),
        password=getattr(settings, "REDIS_PASSWORD", None),
        database=int(getattr(settings, "REDIS_DB", 0)),
    )
enqueue(task_name, payload)

Enqueue a task synchronously via the platform ARQ adapter.

Source code in src/codex_django/messaging/adapters/arq_client.py 
71
72
73
74
def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Enqueue a task synchronously via the platform ARQ adapter."""
    adapter = async_to_sync(self._get_adapter)()
    return adapter.enqueue(task_name, payload)
aenqueue(task_name, payload) async

Enqueue a task asynchronously via the platform ARQ adapter.

Source code in src/codex_django/messaging/adapters/arq_client.py 
76
77
78
79
async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Enqueue a task asynchronously via the platform ARQ adapter."""
    adapter = await self._get_adapter()
    return await adapter.enqueue_async(task_name, payload)

DjangoCacheAdapter

Notification content cache adapter backed by Redis.

Source code in src/codex_django/messaging/adapters/cache_adapter.py 
21
22
23
24
25
26
27
28
29
30
31
32
class DjangoCacheAdapter:
    """Notification content cache adapter backed by Redis."""

    def get(self, key: str) -> str | None:
        """Return a cached content value by key."""
        manager = get_notifications_cache_manager()
        return manager.get(key)

    def set(self, key: str, value: str, timeout: int) -> None:
        """Store a cached content value with a TTL."""
        manager = get_notifications_cache_manager()
        manager.set(key, value, timeout=timeout)
Methods:
get(key)

Return a cached content value by key.

Source code in src/codex_django/messaging/adapters/cache_adapter.py 
24
25
26
27
def get(self, key: str) -> str | None:
    """Return a cached content value by key."""
    manager = get_notifications_cache_manager()
    return manager.get(key)
set(key, value, timeout)

Store a cached content value with a TTL.

Source code in src/codex_django/messaging/adapters/cache_adapter.py 
29
30
31
32
def set(self, key: str, value: str, timeout: int) -> None:
    """Store a cached content value with a TTL."""
    manager = get_notifications_cache_manager()
    manager.set(key, value, timeout=timeout)

DjangoDirectAdapter

Deliver notifications inline via Django send_mail().

Notes

The adapter implements the same enqueue interface as the queue-based adapter, which makes it useful for development, testing, and fallback delivery paths.

Source code in src/codex_django/messaging/adapters/direct_adapter.py 
 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
143
144
145
146
147
148
149
150
151
152
153
class DjangoDirectAdapter:
    """Deliver notifications inline via Django ``send_mail()``.

    Notes:
        The adapter implements the same enqueue interface as the queue-based
        adapter, which makes it useful for development, testing, and fallback
        delivery paths.
    """

    def __init__(
        self,
        renderer: Any = None,
        use_on_commit: bool = True,
    ) -> None:
        self._renderer = renderer
        self._use_on_commit = use_on_commit

    def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
        """Synchronously deliver or schedule delivery of a notification payload."""
        if self._use_on_commit:
            from django.db import transaction

            transaction.on_commit(lambda: self._send(payload))
            return None
        return self._send(payload)

    async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
        """Asynchronously deliver a notification payload in a worker thread."""
        import asyncio

        await asyncio.to_thread(self._send, payload)
        return None

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

    def _send(self, payload: dict[str, Any]) -> str | None:
        """Send a rendered notification payload through Django mail."""
        mode = payload.get("mode", "rendered")

        if mode == "template":
            html_content, text_content = self._render(payload)
        else:
            html_content = payload.get("html_content", "")
            text_content = payload.get("text_content", "")

        recipient = payload.get("recipient_email", "")
        subject = payload.get("subject", "")

        if not recipient:
            log.warning("DjangoDirectAdapter: recipient_email is empty, skipping send")
            return None

        from django.core.mail import send_mail

        from_email = self._resolve_from_email()

        send_mail(
            subject=subject,
            message=text_content,
            from_email=from_email,
            recipient_list=[recipient],
            html_message=html_content or None,
            fail_silently=False,
        )
        log.info("DjangoDirectAdapter: sent email to %s (subject=%r)", recipient, subject)
        return None

    def _resolve_from_email(self) -> str:
        """Resolve From address from SiteSettings (identity mixin) with fallback.

        Priority:
            1. ``SiteSettings.email_from`` (+ ``email_sender_name`` for display)
               resolved via ``CODEX_SITE_SETTINGS_MODEL`` and the Redis-synced
               site-settings manager.
            2. ``settings.DEFAULT_FROM_EMAIL``.
            3. ``"noreply@example.com"``.
        """
        from django.conf import settings

        fallback = getattr(settings, "DEFAULT_FROM_EMAIL", "noreply@example.com")
        model_path = getattr(settings, "CODEX_SITE_SETTINGS_MODEL", None)
        if not model_path:
            return fallback
        try:
            from django.apps import apps

            from codex_django.core.redis.managers.settings import (
                get_site_settings_manager,
            )

            model_cls = apps.get_model(model_path)
            data = get_site_settings_manager().load_cached(model_cls) or {}
        except Exception:  # noqa: BLE001 — site settings are optional
            return fallback

        email = (data.get("email_from") or "").strip()
        if not email:
            return fallback
        name = (data.get("email_sender_name") or "").strip()
        return f"{name} <{email}>" if name else email

    def _render(self, payload: dict[str, Any]) -> tuple[str, str]:
        """Render a template-mode payload using the configured renderer."""
        if self._renderer is None:
            raise ValueError(
                "DjangoDirectAdapter: 'renderer' is required for Mode 1 payloads "
                "(mode='template'). Pass a TemplateRenderer instance or use Mode 2 "
                "(build_rendered) to avoid rendering."
            )
        template_name: str = payload["template_name"]
        context_data: dict[str, Any] = payload.get("context_data", {})
        html_content: str = self._renderer.render(template_name, context_data)
        text_content: str = payload.get("text_content", "")
        return html_content, text_content
Methods:
enqueue(task_name, payload)

Synchronously deliver or schedule delivery of a notification payload.

Source code in src/codex_django/messaging/adapters/direct_adapter.py 
55
56
57
58
59
60
61
62
def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Synchronously deliver or schedule delivery of a notification payload."""
    if self._use_on_commit:
        from django.db import transaction

        transaction.on_commit(lambda: self._send(payload))
        return None
    return self._send(payload)
aenqueue(task_name, payload) async

Asynchronously deliver a notification payload in a worker thread.

Source code in src/codex_django/messaging/adapters/direct_adapter.py 
64
65
66
67
68
69
async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Asynchronously deliver a notification payload in a worker thread."""
    import asyncio

    await asyncio.to_thread(self._send, payload)
    return None

EmailSettingsRedisManager

Bases: BaseDjangoRedisManager

Persist email settings into a dedicated Redis hash.

Source code in src/codex_django/messaging/adapters/email_settings.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
class EmailSettingsRedisManager(BaseDjangoRedisManager):
    """Persist email settings into a dedicated Redis hash."""

    def __init__(self, **kwargs: Any) -> None:
        super().__init__(prefix="", **kwargs)

    def sync(self, data: dict[str, str | None]) -> None:
        """Write the latest serialized email settings to Redis."""
        if self._is_disabled() or not data:
            return
        key = self.make_key(SETTINGS_HASH_KEY)
        with self.sync_hash() as redis_hash:
            redis_hash.set_fields(key, {name: value if value is not None else "" for name, value in data.items()})
Methods:
sync(data)

Write the latest serialized email settings to Redis.

Source code in src/codex_django/messaging/adapters/email_settings.py 
18
19
20
21
22
23
24
def sync(self, data: dict[str, str | None]) -> None:
    """Write the latest serialized email settings to Redis."""
    if self._is_disabled() or not data:
        return
    key = self.make_key(SETTINGS_HASH_KEY)
    with self.sync_hash() as redis_hash:
        redis_hash.set_fields(key, {name: value if value is not None else "" for name, value in data.items()})

DjangoI18nAdapter

Wraps Django's translation.override() for messaging language switching.

Source code in src/codex_django/messaging/adapters/i18n_adapter.py 
18
19
20
21
22
23
24
class DjangoI18nAdapter:
    """Wraps Django's translation.override() for messaging language switching."""

    def translation_override(self, language: str) -> AbstractContextManager[None]:
        from django.utils.translation import override

        return override(language)

DjangoQueueAdapter

Queue-backed notification adapter for ARQ-based delivery.

Source code in src/codex_django/messaging/adapters/queue_adapter.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
class DjangoQueueAdapter:
    """Queue-backed notification adapter for ARQ-based delivery."""

    def __init__(self, arq_client: DjangoArqClient, use_on_commit: bool = True) -> None:
        self._arq = arq_client
        self._use_on_commit = use_on_commit

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

        Returns ``None`` when ``use_on_commit=True`` because the job is not
        enqueued until the transaction commits. Otherwise returns the ARQ job id.
        """
        if self._use_on_commit:
            from django.db import transaction

            transaction.on_commit(lambda: self._arq.enqueue(task_name, payload=payload))
            return None
        return self._arq.enqueue(task_name, payload=payload)

    async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
        """Asynchronously enqueue a notification payload without ``on_commit``."""
        return await self._arq.aenqueue(task_name, payload=payload)
Methods:
enqueue(task_name, payload)

Synchronously enqueue a notification payload.

Returns None when use_on_commit=True because the job is not enqueued until the transaction commits. Otherwise returns the ARQ job id.

Source code in src/codex_django/messaging/adapters/queue_adapter.py 
35
36
37
38
39
40
41
42
43
44
45
46
def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Synchronously enqueue a notification payload.

    Returns ``None`` when ``use_on_commit=True`` because the job is not
    enqueued until the transaction commits. Otherwise returns the ARQ job id.
    """
    if self._use_on_commit:
        from django.db import transaction

        transaction.on_commit(lambda: self._arq.enqueue(task_name, payload=payload))
        return None
    return self._arq.enqueue(task_name, payload=payload)
aenqueue(task_name, payload) async

Asynchronously enqueue a notification payload without on_commit.

Source code in src/codex_django/messaging/adapters/queue_adapter.py 
48
49
50
async def aenqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """Asynchronously enqueue a notification payload without ``on_commit``."""
    return await self._arq.aenqueue(task_name, payload=payload)

BaseAudienceBuilder

Django base for streaming campaign recipients from a configured model.

Source code in src/codex_django/messaging/audience.py 
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
class BaseAudienceBuilder:
    """Django base for streaming campaign recipients from a configured model."""

    recipient_model_setting: str = "MESSAGING_RECIPIENT_MODEL"
    legacy_recipient_model_setting: str = "CONVERSATIONS_RECIPIENT_MODEL"
    chunk_size: int = 500

    def __init__(self) -> None:
        model_label = getattr(
            settings,
            self.recipient_model_setting,
            getattr(settings, self.legacy_recipient_model_setting, ""),
        )
        if not model_label:
            raise AttributeError(
                f"Neither {self.recipient_model_setting} nor {self.legacy_recipient_model_setting} is configured"
            )
        self._model = apps.get_model(model_label)

    def base_queryset(self) -> Any:
        """Return the unfiltered base queryset."""
        return self._model.objects.all()

    def apply_filters(self, qs: Any, audience_filter: dict[str, Any]) -> Any:
        """Project override point for custom queryset filtering."""
        return qs

    def to_draft(self, obj: Any) -> CampaignRecipientDraft:
        """Convert one ORM object into a portable recipient draft."""
        return CampaignRecipientDraft(
            recipient_id=str(obj.pk),
            email=str(getattr(obj, "email", "")),
            first_name=str(getattr(obj, "first_name", "")),
            last_name=str(getattr(obj, "last_name", "")),
            locale=str(getattr(obj, "locale", "de")),
            unsubscribe_token=getattr(obj, "unsubscribe_token", None),
        )

    def count(self, audience_filter: dict[str, Any]) -> int:
        """Count recipients after applying project filters."""
        qs = self.apply_filters(self.base_queryset(), audience_filter)
        return int(qs.count())

    def materialize(self, audience_filter: dict[str, Any]) -> Iterator[CampaignRecipientDraft]:
        """Stream recipient drafts in bounded chunks."""
        qs = self.apply_filters(self.base_queryset(), audience_filter)
        for obj in qs.iterator(chunk_size=self.chunk_size):
            yield self.to_draft(obj)
Methods:
base_queryset()

Return the unfiltered base queryset.

Source code in src/codex_django/messaging/audience.py 
44
45
46
def base_queryset(self) -> Any:
    """Return the unfiltered base queryset."""
    return self._model.objects.all()
apply_filters(qs, audience_filter)

Project override point for custom queryset filtering.

Source code in src/codex_django/messaging/audience.py 
48
49
50
def apply_filters(self, qs: Any, audience_filter: dict[str, Any]) -> Any:
    """Project override point for custom queryset filtering."""
    return qs
to_draft(obj)

Convert one ORM object into a portable recipient draft.

Source code in src/codex_django/messaging/audience.py 
52
53
54
55
56
57
58
59
60
61
def to_draft(self, obj: Any) -> CampaignRecipientDraft:
    """Convert one ORM object into a portable recipient draft."""
    return CampaignRecipientDraft(
        recipient_id=str(obj.pk),
        email=str(getattr(obj, "email", "")),
        first_name=str(getattr(obj, "first_name", "")),
        last_name=str(getattr(obj, "last_name", "")),
        locale=str(getattr(obj, "locale", "de")),
        unsubscribe_token=getattr(obj, "unsubscribe_token", None),
    )
count(audience_filter)

Count recipients after applying project filters.

Source code in src/codex_django/messaging/audience.py 
63
64
65
66
def count(self, audience_filter: dict[str, Any]) -> int:
    """Count recipients after applying project filters."""
    qs = self.apply_filters(self.base_queryset(), audience_filter)
    return int(qs.count())
materialize(audience_filter)

Stream recipient drafts in bounded chunks.

Source code in src/codex_django/messaging/audience.py 
68
69
70
71
72
def materialize(self, audience_filter: dict[str, Any]) -> Iterator[CampaignRecipientDraft]:
    """Stream recipient drafts in bounded chunks."""
    qs = self.apply_filters(self.base_queryset(), audience_filter)
    for obj in qs.iterator(chunk_size=self.chunk_size):
        yield self.to_draft(obj)

CampaignRecipientDraft dataclass

Portable recipient row for campaign batching.

Source code in src/codex_django/messaging/audience.py 
13
14
15
16
17
18
19
20
21
22
@dataclass(frozen=True)
class CampaignRecipientDraft:
    """Portable recipient row for campaign batching."""

    recipient_id: str
    email: str
    first_name: str = ""
    last_name: str = ""
    locale: str = "de"
    unsubscribe_token: str | None = None

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.

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 {},
    }

MessagingBridge

Bases: Protocol

Project adapter that exposes messaging state and write actions to cabinet.

Source code in src/codex_django/messaging/cabinet/bridge.py 
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
class MessagingBridge(Protocol):
    """Project adapter that exposes messaging state and write actions to cabinet."""

    def get_mailbox_state(
        self,
        *,
        user: Any,
        folder: str = "inbox",
        thread_id: str | None = None,
        filters: Mapping[str, Any] | None = None,
    ) -> MailboxState: ...

    def get_campaign_list_state(
        self,
        *,
        user: Any,
        status: str = "all",
        filters: Mapping[str, Any] | None = None,
    ) -> CampaignListState: ...

    def get_campaign_detail_state(self, *, user: Any, campaign_id: str) -> CampaignDetailState: ...

    def get_campaign_composer_state(
        self,
        *,
        user: Any,
        campaign_id: str | None = None,
    ) -> CampaignComposerState: ...

    def get_recipient_list_state(
        self,
        *,
        user: Any,
        filters: Mapping[str, Any] | None = None,
    ) -> RecipientListState: ...

    def get_delivery_log_state(
        self,
        *,
        user: Any,
        filters: Mapping[str, Any] | None = None,
    ) -> DeliveryLogState: ...

    def get_settings_state(self, *, user: Any) -> MessagingSettingsState: ...

    def reply_to_thread(self, *, user: Any, thread_id: str, body: str) -> MessagingActionResult: ...

    def mark_thread_read(self, *, user: Any, thread_id: str) -> MessagingActionResult: ...

    def mark_thread_processed(self, *, user: Any, thread_id: str) -> MessagingActionResult: ...

    def mark_thread_spam(self, *, user: Any, thread_id: str) -> MessagingActionResult: ...

    def archive_thread(self, *, user: Any, thread_id: str) -> MessagingActionResult: ...

    def create_campaign(self, *, user: Any, payload: Mapping[str, Any]) -> MessagingActionResult: ...

    def update_campaign(self, *, user: Any, campaign_id: str, payload: Mapping[str, Any]) -> MessagingActionResult: ...

    def send_campaign(self, *, user: Any, campaign_id: str) -> MessagingActionResult: ...

    def send_test_campaign(self, *, user: Any, campaign_id: str, email: str) -> MessagingActionResult: ...

    def update_recipient(
        self, *, user: Any, recipient_id: str, payload: Mapping[str, Any]
    ) -> MessagingActionResult: ...

    def update_settings(self, *, user: Any, payload: Mapping[str, Any]) -> MessagingActionResult: ...

MessagingCabinetPresenter

Map project-provided messaging states into cabinet template contexts.

Source code in src/codex_django/messaging/cabinet/presenters.py 
 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
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
class MessagingCabinetPresenter:
    """Map project-provided messaging states into cabinet template contexts."""

    def mailbox_context(self, state: MailboxState) -> dict[str, Any]:
        return {
            "mailbox": state,
            "mailbox_panel": SplitPanelData(
                items=[self._thread_row(thread) for thread in state.threads],
                active_id=state.selected.thread.id if state.selected and state.selected.thread else "",
                detail_url="",
                empty_message=state.empty_message,
            ),
            "unread_messages_count": sum(1 for thread in state.threads if thread.unread),
        }

    def campaign_list_context(self, state: CampaignListState) -> dict[str, Any]:
        return {"campaigns": state, "campaigns_table": self._campaign_table(state)}

    def campaign_composer_context(self, state: CampaignComposerState) -> dict[str, Any]:
        return {"campaign": state}

    def campaign_detail_context(self, state: CampaignDetailState) -> dict[str, Any]:
        return {"campaign": state}

    def recipient_list_context(self, state: RecipientListState) -> dict[str, Any]:
        return {"recipients": state, "recipients_table": self._recipient_table(state)}

    def delivery_log_context(self, state: DeliveryLogState) -> dict[str, Any]:
        return {"delivery_log": state, "delivery_log_table": self._delivery_table(state)}

    def settings_context(self, state: MessagingSettingsState) -> dict[str, Any]:
        return {"messaging_settings": state}

    @staticmethod
    def _thread_row(thread: MailboxThreadState) -> ListRow:
        secondary_parts = [part for part in [thread.sender_label or thread.sender_name, thread.preview] if part]
        return ListRow(
            id=thread.id,
            primary=thread.subject,
            secondary=" - ".join(secondary_parts),
            meta=thread.received_at,
            avatar=(thread.sender_label or thread.sender_name or thread.sender_email)[:2].upper(),
            url=thread.url,
        )

    @staticmethod
    def _campaign_row(row: CampaignRowState) -> dict[str, Any]:
        status_label = row.status_label or row.status
        return {
            "id": row.id,
            "subject": row.subject,
            "status": row.status,
            "status_label": status_label,
            "locale": row.locale,
            "recipient_count": row.recipient_count,
            "sent_count": row.sent_count,
            "failed_count": row.failed_count,
            "scheduled_at": row.scheduled_at,
            "sent_at": row.sent_at,
            "updated_at": row.updated_at,
            "url": row.url,
            "status_colors": {
                "draft": "secondary",
                "scheduled": "info",
                "sending": "primary",
                "sent": "success",
                "failed": "danger",
                "cancelled": "dark",
            },
        }

    def _campaign_table(self, state: CampaignListState) -> DataTableData:
        return DataTableData(
            columns=[
                TableColumn(key="subject", label="Subject", bold=True),
                TableColumn(key="status_label", label="Status", badge_key="status_colors"),
                TableColumn(key="locale", label="Locale"),
                TableColumn(key="recipient_count", label="Recipients", align="right"),
                TableColumn(key="sent_count", label="Sent", align="right"),
                TableColumn(key="failed_count", label="Failed", align="right"),
                TableColumn(key="updated_at", label="Updated", muted=True),
            ],
            rows=[self._campaign_row(row) for row in state.rows],
            filters=[TableFilter(key=item.key, label=item.label, value=item.key) for item in state.status_filters],
            actions=[TableAction(label="Open", url_key="url", icon="bi-eye")],
            search_placeholder="Search campaigns...",
            empty_message=state.empty_message,
        )

    @staticmethod
    def _recipient_row(row: RecipientRowState) -> dict[str, Any]:
        status = "enabled" if row.enabled else "disabled"
        return {
            "id": row.id,
            "email": row.email,
            "name": row.name,
            "kind": row.kind,
            "locale": row.locale,
            "status": status,
            "enabled_label": "Enabled" if row.enabled else "Disabled",
            "note": row.note,
            "url": row.url,
            "status_colors": {"enabled": "success", "disabled": "secondary"},
        }

    def _recipient_table(self, state: RecipientListState) -> DataTableData:
        return DataTableData(
            columns=[
                TableColumn(key="email", label="Email", bold=True),
                TableColumn(key="name", label="Name"),
                TableColumn(key="kind", label="Kind"),
                TableColumn(key="locale", label="Locale"),
                TableColumn(key="enabled_label", label="Status", badge_key="status_colors"),
                TableColumn(key="note", label="Note", muted=True),
            ],
            rows=[self._recipient_row(row) for row in state.rows],
            filters=list(state.filters),
            actions=[TableAction(label="Open", url_key="url", icon="bi-eye")],
            search_placeholder="Search recipients...",
            empty_message=state.empty_message,
        )

    @staticmethod
    def _delivery_row(row: DeliveryLogRowState) -> dict[str, Any]:
        return {
            "id": row.id,
            "recipient": row.recipient,
            "event_type": row.event_type,
            "channel": row.channel,
            "status": row.status,
            "subject": row.subject,
            "queued_at": row.queued_at,
            "sent_at": row.sent_at,
            "error_message": row.error_message,
            "provider_message_id": row.provider_message_id,
            "status_colors": {
                "queued": "secondary",
                "sent": "success",
                "failed": "danger",
                "bounced": "warning",
                "skipped": "dark",
            },
        }

    def _delivery_table(self, state: DeliveryLogState) -> DataTableData:
        return DataTableData(
            columns=[
                TableColumn(key="recipient", label="Recipient", bold=True),
                TableColumn(key="event_type", label="Event"),
                TableColumn(key="channel", label="Channel"),
                TableColumn(key="status", label="Status", badge_key="status_colors"),
                TableColumn(key="subject", label="Subject"),
                TableColumn(key="sent_at", label="Sent at", muted=True),
                TableColumn(key="error_message", label="Error", muted=True),
            ],
            rows=[self._delivery_row(row) for row in state.rows],
            filters=list(state.filters),
            search_placeholder="Search delivery log...",
            empty_message=state.empty_message,
        )

MessagingCabinetWorkflowService

Validate simple cabinet payloads and delegate writes to a project bridge.

Source code in src/codex_django/messaging/cabinet/workflows.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
class MessagingCabinetWorkflowService:
    """Validate simple cabinet payloads and delegate writes to a project bridge."""

    def __init__(self, bridge: MessagingBridge) -> None:
        self.bridge = bridge

    def reply(self, *, user: Any, thread_id: str, payload: Mapping[str, Any]) -> MessagingActionResult:
        body = self._string(payload.get("body"))
        if not body:
            return self._invalid("reply.invalid", "Reply body is required.", {"body": ["Reply body is required."]})
        return self.bridge.reply_to_thread(user=user, thread_id=thread_id, body=body)

    def mark_read(self, *, user: Any, thread_id: str) -> MessagingActionResult:
        return self.bridge.mark_thread_read(user=user, thread_id=thread_id)

    def mark_processed(self, *, user: Any, thread_id: str) -> MessagingActionResult:
        return self.bridge.mark_thread_processed(user=user, thread_id=thread_id)

    def mark_spam(self, *, user: Any, thread_id: str) -> MessagingActionResult:
        return self.bridge.mark_thread_spam(user=user, thread_id=thread_id)

    def archive(self, *, user: Any, thread_id: str) -> MessagingActionResult:
        return self.bridge.archive_thread(user=user, thread_id=thread_id)

    def save_campaign(
        self,
        *,
        user: Any,
        payload: Mapping[str, Any],
        campaign_id: str | None = None,
    ) -> MessagingActionResult:
        errors: dict[str, list[str]] = {}
        if not self._string(payload.get("subject")):
            errors["subject"] = ["Subject is required."]
        if not self._string(payload.get("body_text") or payload.get("body")):
            errors["body_text"] = ["Message body is required."]
        if errors:
            return self._invalid("campaign.invalid", "Campaign payload is invalid.", errors)
        if campaign_id:
            return self.bridge.update_campaign(user=user, campaign_id=campaign_id, payload=payload)
        return self.bridge.create_campaign(user=user, payload=payload)

    def send_campaign(self, *, user: Any, campaign_id: str) -> MessagingActionResult:
        return self.bridge.send_campaign(user=user, campaign_id=campaign_id)

    def send_test_campaign(self, *, user: Any, campaign_id: str, payload: Mapping[str, Any]) -> MessagingActionResult:
        email = self._string(payload.get("email"))
        if not email or "@" not in email:
            return self._invalid("campaign_test.invalid", "A valid email is required.", {"email": ["Enter an email."]})
        return self.bridge.send_test_campaign(user=user, campaign_id=campaign_id, email=email)

    def save_recipient(self, *, user: Any, recipient_id: str, payload: Mapping[str, Any]) -> MessagingActionResult:
        email = self._string(payload.get("email"))
        if not email or "@" not in email:
            return self._invalid("recipient.invalid", "A valid email is required.", {"email": ["Enter an email."]})
        return self.bridge.update_recipient(user=user, recipient_id=recipient_id, payload=payload)

    def save_settings(self, *, user: Any, payload: Mapping[str, Any]) -> MessagingActionResult:
        if not payload:
            return self._invalid(
                "settings.invalid",
                "Settings payload is empty.",
                {"__all__": ["No settings provided."]},
            )
        return self.bridge.update_settings(user=user, payload=payload)

    def dispatch(
        self,
        action: str,
        *,
        user: Any,
        payload: Mapping[str, Any] | None = None,
        thread_id: str = "",
        campaign_id: str = "",
        recipient_id: str = "",
    ) -> MessagingActionResult:
        payload = payload or {}
        if action == "reply":
            return self.reply(user=user, thread_id=thread_id, payload=payload)
        if action == "mark_read":
            return self.mark_read(user=user, thread_id=thread_id)
        if action == "mark_processed":
            return self.mark_processed(user=user, thread_id=thread_id)
        if action == "mark_spam":
            return self.mark_spam(user=user, thread_id=thread_id)
        if action == "archive":
            return self.archive(user=user, thread_id=thread_id)
        if action == "save_campaign":
            return self.save_campaign(user=user, campaign_id=campaign_id or None, payload=payload)
        if action == "send_campaign":
            return self.send_campaign(user=user, campaign_id=campaign_id)
        if action == "send_test_campaign":
            return self.send_test_campaign(user=user, campaign_id=campaign_id, payload=payload)
        if action == "save_recipient":
            return self.save_recipient(user=user, recipient_id=recipient_id, payload=payload)
        if action == "save_settings":
            return self.save_settings(user=user, payload=payload)
        return MessagingActionResult(ok=False, code="unknown_action", message=f"Unknown messaging action: {action}")

    @staticmethod
    def _string(value: Any) -> str:
        return str(value or "").strip()

    @staticmethod
    def _invalid(code: str, message: str, field_errors: dict[str, list[str]]) -> MessagingActionResult:
        return MessagingActionResult(ok=False, code=code, message=message, field_errors=field_errors)

CampaignBatch dataclass

Bounded batch handed to a project-owned dispatcher.

Source code in src/codex_django/messaging/campaigns.py 
12
13
14
15
16
17
18
19
20
21
22
23
24
@dataclass(frozen=True)
class CampaignBatch:
    """Bounded batch handed to a project-owned dispatcher."""

    campaign_id: str
    recipients: list[CampaignRecipientDraft]
    subject: str
    template_key: str = ""
    locale: str = "de"
    body_text: str = ""
    body_translations: dict[str, str] = field(default_factory=dict)
    audience_filter: dict[str, Any] = field(default_factory=dict)
    base_context: dict[str, Any] = field(default_factory=dict)

CampaignDispatcherProtocol

Bases: Protocol

Minimal dispatcher interface for campaign batch handoff.

Source code in src/codex_django/messaging/campaigns.py 
27
28
29
30
class CampaignDispatcherProtocol(Protocol):
    """Minimal dispatcher interface for campaign batch handoff."""

    def enqueue_batch(self, batch: CampaignBatch) -> str: ...

CampaignService

Compose audience streaming and dispatcher handoff for campaigns.

Source code in src/codex_django/messaging/campaigns.py 
 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
class CampaignService:
    """Compose audience streaming and dispatcher handoff for campaigns."""

    batch_size: int = 25

    def __init__(
        self,
        *,
        audience: BaseAudienceBuilder,
        dispatcher: CampaignDispatcherProtocol,
        locales: Iterable[str] | None = None,
        batch_size: int | None = None,
    ) -> None:
        self._audience = audience
        self._dispatcher = dispatcher
        self._locales = list(locales or ["de"])
        if batch_size is not None:
            self.batch_size = batch_size

    def count_recipients(self, audience_filter: dict[str, Any]) -> int:
        """Return the filtered audience size."""
        return self._audience.count(audience_filter)

    def build_batches(
        self,
        *,
        campaign_id: str,
        subject: str,
        audience_filter: dict[str, Any],
        template_key: str = "",
        locale: str = "de",
        body_text: str = "",
        body_translations: dict[str, str] | None = None,
        base_context: dict[str, Any] | None = None,
    ) -> list[CampaignBatch]:
        """Materialize the audience into bounded dispatcher batches."""
        current: list[CampaignRecipientDraft] = []
        batches: list[CampaignBatch] = []

        for recipient in self._audience.materialize(audience_filter):
            current.append(recipient)
            if len(current) >= self.batch_size:
                batches.append(
                    CampaignBatch(
                        campaign_id=campaign_id,
                        recipients=list(current),
                        subject=subject,
                        template_key=template_key,
                        locale=locale,
                        body_text=body_text,
                        body_translations=dict(body_translations or {}),
                        audience_filter=dict(audience_filter),
                        base_context=dict(base_context or {}),
                    )
                )
                current.clear()

        if current:
            batches.append(
                CampaignBatch(
                    campaign_id=campaign_id,
                    recipients=list(current),
                    subject=subject,
                    template_key=template_key,
                    locale=locale,
                    body_text=body_text,
                    body_translations=dict(body_translations or {}),
                    audience_filter=dict(audience_filter),
                    base_context=dict(base_context or {}),
                )
            )

        return batches

    def send(
        self,
        *,
        campaign_id: str,
        subject: str,
        audience_filter: dict[str, Any],
        template_key: str = "",
        locale: str = "de",
        body_text: str = "",
        body_translations: dict[str, str] | None = None,
        base_context: dict[str, Any] | None = None,
    ) -> list[str]:
        """Build and enqueue all campaign batches."""
        job_ids: list[str] = []
        for batch in self.build_batches(
            campaign_id=campaign_id,
            subject=subject,
            audience_filter=audience_filter,
            template_key=template_key,
            locale=locale,
            body_text=body_text,
            body_translations=body_translations,
            base_context=base_context,
        ):
            job_ids.append(self._dispatcher.enqueue_batch(batch))
        return job_ids
Methods:
count_recipients(audience_filter)

Return the filtered audience size.

Source code in src/codex_django/messaging/campaigns.py 
52
53
54
def count_recipients(self, audience_filter: dict[str, Any]) -> int:
    """Return the filtered audience size."""
    return self._audience.count(audience_filter)
build_batches(*, campaign_id, subject, audience_filter, template_key='', locale='de', body_text='', body_translations=None, base_context=None)

Materialize the audience into bounded dispatcher batches.

Source code in src/codex_django/messaging/campaigns.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
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def build_batches(
    self,
    *,
    campaign_id: str,
    subject: str,
    audience_filter: dict[str, Any],
    template_key: str = "",
    locale: str = "de",
    body_text: str = "",
    body_translations: dict[str, str] | None = None,
    base_context: dict[str, Any] | None = None,
) -> list[CampaignBatch]:
    """Materialize the audience into bounded dispatcher batches."""
    current: list[CampaignRecipientDraft] = []
    batches: list[CampaignBatch] = []

    for recipient in self._audience.materialize(audience_filter):
        current.append(recipient)
        if len(current) >= self.batch_size:
            batches.append(
                CampaignBatch(
                    campaign_id=campaign_id,
                    recipients=list(current),
                    subject=subject,
                    template_key=template_key,
                    locale=locale,
                    body_text=body_text,
                    body_translations=dict(body_translations or {}),
                    audience_filter=dict(audience_filter),
                    base_context=dict(base_context or {}),
                )
            )
            current.clear()

    if current:
        batches.append(
            CampaignBatch(
                campaign_id=campaign_id,
                recipients=list(current),
                subject=subject,
                template_key=template_key,
                locale=locale,
                body_text=body_text,
                body_translations=dict(body_translations or {}),
                audience_filter=dict(audience_filter),
                base_context=dict(base_context or {}),
            )
        )

    return batches
send(*, campaign_id, subject, audience_filter, template_key='', locale='de', body_text='', body_translations=None, base_context=None)

Build and enqueue all campaign batches.

Source code in src/codex_django/messaging/campaigns.py 
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
def send(
    self,
    *,
    campaign_id: str,
    subject: str,
    audience_filter: dict[str, Any],
    template_key: str = "",
    locale: str = "de",
    body_text: str = "",
    body_translations: dict[str, str] | None = None,
    base_context: dict[str, Any] | None = None,
) -> list[str]:
    """Build and enqueue all campaign batches."""
    job_ids: list[str] = []
    for batch in self.build_batches(
        campaign_id=campaign_id,
        subject=subject,
        audience_filter=audience_filter,
        template_key=template_key,
        locale=locale,
        body_text=body_text,
        body_translations=body_translations,
        base_context=base_context,
    ):
        job_ids.append(self._dispatcher.enqueue_batch(batch))
    return job_ids

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: ...

MessagingEventRegistry

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

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)

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.

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

Functions:

get_email_settings_manager()

Return the default email settings Redis manager.

Source code in src/codex_django/messaging/adapters/email_settings.py 
27
28
29
def get_email_settings_manager() -> EmailSettingsRedisManager:
    """Return the default email settings Redis manager."""
    return EmailSettingsRedisManager()

build_default_mailbox_folders(*, base_url, active_folder='inbox')

Return common mailbox folders that projects may override in their bridge.

Source code in src/codex_django/messaging/cabinet/types/mailbox.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
def build_default_mailbox_folders(*, base_url: str, active_folder: str = "inbox") -> list[MailboxFolderState]:
    """Return common mailbox folders that projects may override in their bridge."""
    definitions = [
        ("inbox", "Inbox", "bi-inbox"),
        ("unread", "Unread", "bi-envelope"),
        ("processed", "Processed", "bi-check2-circle"),
        ("spam", "Spam", "bi-exclamation-octagon"),
        ("archived", "Archived", "bi-archive"),
        ("all", "All", "bi-collection"),
    ]
    separator = "&" if "?" in base_url else "?"
    return [
        MailboxFolderState(
            key=key,
            label=label,
            icon=icon,
            url=f"{base_url}{separator}folder={key}",
            active=key == active_folder,
        )
        for key, label, icon in definitions
    ]

build_messaging_sidebar(*, mailbox_url, campaigns_url, recipients_url, delivery_log_url, unread_badge_key='unread_messages_count')

Return the default Messaging sidebar without registering it globally.

Source code in src/codex_django/messaging/cabinet/navigation.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def build_messaging_sidebar(
    *,
    mailbox_url: str,
    campaigns_url: str,
    recipients_url: str,
    delivery_log_url: str,
    unread_badge_key: str = "unread_messages_count",
) -> list[SidebarItem]:
    """Return the default Messaging sidebar without registering it globally."""
    return [
        SidebarItem(label="Mailbox", url=mailbox_url, icon="bi-inbox", badge_key=unread_badge_key, order=10),
        SidebarItem(label="Campaigns", url=campaigns_url, icon="bi-megaphone", order=20),
        SidebarItem(label="Recipients", url=recipients_url, icon="bi-people", order=30),
        SidebarItem(label="Delivery log", url=delivery_log_url, icon="bi-list-check", order=40),
    ]

build_messaging_topbar(url, *, label='Messaging', group='services', order=20)

Return the default Messaging topbar entry without registering it globally.

Source code in src/codex_django/messaging/cabinet/navigation.py 
25
26
27
28
29
30
31
32
33
def build_messaging_topbar(
    url: str,
    *,
    label: str = "Messaging",
    group: str = "services",
    order: int = 20,
) -> TopbarEntry:
    """Return the default Messaging topbar entry without registering it globally."""
    return TopbarEntry(group=group, label=label, icon="bi-envelope-paper", url=url, order=order)

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)