Skip to content

notifications.delivery.base

base

codex_platform.notifications.delivery.base

Protocol definition for notification delivery adapters.

Classes

NotificationAdapter

Bases: Protocol

Contract for notification delivery transport.

Allows the same business logic to work with ARQ, Celery, Direct calls, or Django's built-in mail system.

Implementations MUST
  • Raise exceptions on infrastructure failures (network, broker, DB).
  • Return a job/task ID (str) when the backend provides one.
  • Return None for fire-and-forget transports with no tracking ID.
Source code in src/codex_platform/notifications/delivery/base.py 
10
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
class NotificationAdapter(Protocol):
    """
    Contract for notification delivery transport.

    Allows the same business logic to work with ARQ, Celery, Direct calls,
    or Django's built-in mail system.

    Implementations MUST:
        - Raise exceptions on infrastructure failures (network, broker, DB).
        - Return a job/task ID (str) when the backend provides one.
        - Return None for fire-and-forget transports with no tracking ID.
    """

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

        Args:
            task_name: Worker function name to execute.
                       May be unused by adapters that deliver synchronously.
            payload: Serialized ``NotificationPayloadDTO`` (via ``.model_dump(mode="json")``).

        Returns:
            str: Job/task identifier for tracking.
            None: When the transport is fire-and-forget.

        Raises:
            Exception: Infrastructure errors MUST propagate — never swallow them.
        """
        ...
Functions
enqueue(task_name, payload)

Deliver or enqueue the notification.

Parameters:

Name Type Description Default
task_name str

Worker function name to execute. May be unused by adapters that deliver synchronously.

 required
payload dict[str, Any]

Serialized NotificationPayloadDTO (via .model_dump(mode="json")).

 required

Returns:

Name Type Description
str str | None

Job/task identifier for tracking.
None str | None

When the transport is fire-and-forget.

Raises:

Type Description
Exception

Infrastructure errors MUST propagate — never swallow them.
Source code in src/codex_platform/notifications/delivery/base.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
def enqueue(self, task_name: str, payload: dict[str, Any]) -> str | None:
    """
    Deliver or enqueue the notification.

    Args:
        task_name: Worker function name to execute.
                   May be unused by adapters that deliver synchronously.
        payload: Serialized ``NotificationPayloadDTO`` (via ``.model_dump(mode="json")``).

    Returns:
        str: Job/task identifier for tracking.
        None: When the transport is fire-and-forget.

    Raises:
        Exception: Infrastructure errors MUST propagate — never swallow them.
    """
    ...