Notifications¶

Introduction¶

In addition to mail, Arvel supports sending notifications across a variety of channels — log, mail, database, and broadcast. A single notification class describes the message once, then renders itself differently per channel. Notifications are sent to notifiables (typically your User model).

Registering the Provider¶

Notifications are opt-in. Add NotificationServiceProvider to bootstrap/providers.py. It binds the Notification facade.

The available channels depend on what else is bound: the log and broadcast channels are always available; the mail channel registers only when the Mailer is bound; the database channel registers only when a SQLAlchemy session factory is bound.

Generating Notifications¶

A notification subclasses Notification and implements via() to declare its channels, plus one to_* method per channel. Place them under app/notifications/.

from typing import Any
from arvel.notifications.notification import Notification
from arvel.mail.mailable import Mailable


class InvoicePaid(Notification):
    def __init__(self, invoice_id: int) -> None:
        self.invoice_id = invoice_id

    def via(self, notifiable: Any) -> list[str]:
        return ["mail", "database"]

    def to_mail(self, notifiable: Any) -> Mailable | None:
        return InvoicePaidMail(self.invoice_id)

    def to_database(self, notifiable: Any) -> dict[str, Any]:
        return {"invoice_id": self.invoice_id}

InvoicePaidMail is one of your own mailables — to_mail() returns it (or None to skip the mail channel).

Specifying Delivery Channels¶

via() returns the list of channel names to use for a given notifiable. You can branch on the notifiable to choose channels per user:

def via(self, notifiable: Any) -> list[str]:
    return notifiable.notification_channels or ["mail"]

Channel Formatting¶

Mail Notifications¶

to_mail() returns a Mailable (or None to skip). The mail channel delivers it through the bound Mailer.

Database Notifications¶

to_database() returns a dict stored in the notifications table's data column. The default is an empty dict.

Broadcast Notifications¶

to_broadcast() returns a dict pushed through the broadcast driver. For the broadcast channel the dict must carry channels and data — e.g. {"channels": ["users.1"], "data": {"invoice_id": 42}}. A payload missing those keys is logged and skipped.

Sending Notifications¶

Send through the Notification facade. Both methods are coroutines:

from arvel.facades.notification import Notification

await Notification.send(user, InvoicePaid(invoice_id=42))

If a single channel fails, the error is logged and the remaining channels still deliver.

Queued Notifications¶

Mix in ShouldQueue to push delivery onto the queue instead of sending inline:

from arvel.notifications.notification import Notification
from arvel.notifications.should_queue import ShouldQueue


class InvoicePaid(Notification, ShouldQueue):
    ...

Notification.send() enqueues queued notifications automatically. Use Notification.send_now() to force inline delivery and bypass the queue. If the queue isn't configured, queued notifications fall back to inline delivery.