Skip to content

Notifications

Notifications provide a unified interface for sending messages across multiple channels—email, SMS, push notifications, Slack, etc. Define notification classes with channel-specific formatting and let the framework handle delivery via the queue.

Making models notifiable

Models that receive notifications need the @notifiable decorator:

from fast_app import Model
from fast_app.decorators.model_decorators import notifiable


@notifiable
class User(Model):
    name: str
    email: str | None = None
    phone: str | None = None

The decorator adds notification helpers to the model (currently just marks it as notifiable; channels determine routing logic).

Generating notifications and channels

Use the CLI to scaffold notification and channel classes:

fast-app make notification EmailOTP
fast-app make notification_channel MailChannel

Notifications go to app/notifications/, channels to app/notification_channels/.

For common channels (mail, Telegram), publish pre-built templates:

fast-app publish mail_notification_channel
fast-app publish telegram_notification_channel

Creating a notification

Notifications extend fast_app.contracts.notification.Notification and define:

  • via(notifiable) — returns a list of channel instances
  • Channel-specific methods like to_mail(notifiable), to_sms(notifiable), etc.
from typing import TYPE_CHECKING

from fast_app import Notification
from fast_app.integrations.notifications.mail import MailMessage
from app.notification_channels.mail_channel import MailChannel

if TYPE_CHECKING:
    from fast_app import Model, NotificationChannel
    from app.models.email_otp import EmailOTP


class EmailOTPNotification(Notification):
    def __init__(self, email_otp: "EmailOTP"):
        self.email_otp = email_otp

    def via(self, notifiable: "Model") -> list["NotificationChannel"]:
        return [MailChannel()]

    async def to_mail(self, notifiable: "Model") -> MailMessage:
        otp = str(self.email_otp.otp)
        otp = otp[:3] + "-" + otp[3:] if len(otp) > 3 else otp

        return MailMessage(
            subject=f"🔑 {otp} is your verification code",
            body=f"Your code: {otp}. Expires in 15 minutes.",
        )

The notification class holds state (e.g., email_otp) and transforms it into channel-specific payloads.

Creating a channel

Channels extend fast_app.contracts.notification_channel.NotificationChannel and implement send(notifiable, notification):

from typing import TYPE_CHECKING

from fast_app import NotificationChannel
from fast_app.integrations.notifications.mail import Mail

if TYPE_CHECKING:
    from app.models.user import User
    from fast_app import Notification


class MailChannel(NotificationChannel):
    async def send(self, notifiable: "User", notification: "Notification"):
        if receiver_mail := notifiable.get("email"):
            await Mail.send(receiver_mail, await notification.to_mail(notifiable))

Channels: 1. Extract recipient details from the notifiable model (e.g., user.email) 2. Call the notification's channel-specific method (to_mail) 3. Send via the integration (e.g., Mail.send)

Sending notifications

Call await notification.send(notifiable) from anywhere in your application:

from app.notifications.email_otp_notification import EmailOTPNotification

otp = await EmailOTP.create({"user_id": user.id, "otp": "123456"})
await EmailOTPNotification(otp).send(user)

Under the hood, send(): 1. Calls via(notifiable) to get channels 2. Enqueues channel.send(notifiable, self) for each channel via fast_app.core.queue.queue 3. Workers process deliveries asynchronously (when QUEUE_DRIVER=async_farm)

Built-in integrations

FastApp ships with helpers for common channels:

Mail

from fast_app.integrations.notifications.mail import Mail, MailMessage

message = MailMessage(
    subject="Welcome!",
    body="Thanks for signing up.",
    html="<p>Thanks for signing up.</p>",
)
await Mail.send("user@example.com", message)

Configure via environment variables: MAIL_HOST, MAIL_PORT, MAIL_USERNAME, MAIL_PASSWORD, MAIL_FROM.

Expo Push Notifications

from fast_app.integrations.notifications.expo import Expo

await Expo.send_push_notification(
    tokens=["ExponentPushToken[...]"],
    title="New message",
    body="You have a new chat message",
)

Requires EXPO_ACCESS_TOKEN.

Slack

from fast_app.integrations.notifications.slack import Slack

await Slack.send_message(
    channel="#alerts",
    text="Deployment completed successfully",
)

Configure via SLACK_WEBHOOK_URL.

Custom channels

Create a custom channel for any service:

from fast_app import NotificationChannel


class SMSChannel(NotificationChannel):
    async def send(self, notifiable, notification):
        phone = notifiable.get("phone")
        message = await notification.to_sms(notifiable)
        await twilio_client.messages.create(to=phone, body=message)

Add a to_sms() method to your notifications:

class EmailOTPNotification(Notification):
    def via(self, notifiable):
        channels = [MailChannel()]
        if notifiable.get("phone"):
            channels.append(SMSChannel())
        return channels

    async def to_sms(self, notifiable):
        return f"Your verification code: {self.email_otp.otp}"

Tips

  • Use via(notifiable) to conditionally select channels based on user preferences or availability (e.g., only send SMS if phone is present).
  • Keep notification classes stateless except for the data they carry; channels handle delivery logic.
  • Test notifications by calling notification.to_mail(user) directly without invoking send().
  • Combine with events: dispatch an event that triggers a listener which sends a notification.

Notifications centralize message formatting and delivery, making it easy to support new channels or update content without touching controllers.