Skip to content

Notifications

Craft Easy ships a multi-channel notification system covering email, SMS, and push notifications. It is provider-based: any channel can be served by one or more NotificationProvider implementations, and the registry transparently handles retries, fallbacks, and dry-run mode.

The built-in providers are:

Channel Provider Module
Email SendGrid craft_easy.core.notifications.providers.sendgrid
SMS 46elks craft_easy.core.notifications.providers.fortysix_elks
Push Firebase Cloud Messaging (FCM) craft_easy.core.notifications.providers.fcm
Push Apple Push Notifications (APNs) craft_easy.core.notifications.providers.apns
Any Console (development only) craft_easy.core.notifications.providers.console

You can register custom providers by subclassing NotificationProvider.

Configuration

All settings live in settings.py and are loaded from environment variables. A minimum email + SMS + push configuration looks like:

# Email — SendGrid
SENDGRID_API_KEY=SG.xxx
SENDGRID_FROM_EMAIL=noreply@example.com
SENDGRID_FROM_NAME="Acme Parking"

# SMS — 46elks
ELKS_API_USERNAME=u...
ELKS_API_PASSWORD=p...
ELKS_SENDER="AcmePark"

# Push — Firebase
FCM_PROJECT_ID=my-project
FCM_SERVICE_ACCOUNT_JSON='{"type":"service_account", ...}'

# Development only — log notifications instead of sending
NOTIFICATION_DRY_RUN=false

In development, if no credentials are set, the setup helper wires the Console provider for every channel and logs every send to stdout.

Sending a notification

NotificationRegistry.send() is the single entry point. It accepts the channel, recipient, subject/body, and any provider-specific **kwargs.

from craft_easy.core.notifications import (
    NotificationChannel,
    get_notification_registry,
)

registry = get_notification_registry()

# Email
result = await registry.send(
    channel=NotificationChannel.EMAIL,
    to="driver@example.com",
    subject="Your parking receipt",
    body="Thank you for your payment.",
    html="<p>Thank you for your payment.</p>",
)

# SMS
result = await registry.send(
    channel=NotificationChannel.SMS,
    to="+46701234567",
    body="Your access code is 4711.",
)

# Push (FCM)
result = await registry.send(
    channel=NotificationChannel.PUSH,
    to="device-token-here",
    body="Your parking session starts now.",
    title="Acme Parking",
    data={"session_id": "abc123"},
)

if not result.success:
    logger.error("Notification failed: %s", result.error)

NotificationResult contains everything you need for logging:

@dataclass
class NotificationResult:
    success: bool
    provider: str
    channel: NotificationChannel
    recipient: str
    message_id: str | None   # Provider-side ID
    error: str | None
    attempts: int            # Total attempts across retries
    metadata: dict

Retries and fallbacks

registry.send(..., max_retries=3) retries the primary provider up to max_retries times before switching to the first registered fallback. Only transient errors trigger retries — permanent errors (400 Bad Request, 401 Unauthorized) are returned immediately so you can fix the input instead of hammering the provider.

If every primary and fallback provider fails, the registry returns a NotificationResult with success=False and error set to the last error seen.

Provider health checks

Each provider exposes async is_healthy() -> bool, which does a cheap probe (GET /scopes for SendGrid, GET /a1/me for 46elks, etc.). Use this to implement /health/notifications or to skip a provider in your own degraded-mode fallback logic.

Notification templates

Hard-coded strings are fine for one-off internal emails, but user-facing notifications should go through the template system. Templates are stored as NotificationTemplate documents and rendered with Jinja2:

from craft_easy.core.notifications.templates import render_notification

body, subject = await render_notification(
    name="otp_email",
    channel="email",
    variables={"code": "847362", "minutes": 10},
    locale="sv",
)

await registry.send(
    channel=NotificationChannel.EMAIL,
    to="user@example.com",
    subject=subject,
    body=body,
)

The NotificationTemplate model

class NotificationTemplate(BaseDocument):
    name: str                        # unique, e.g. "otp_email"
    description: str | None
    channel: str                     # "email" | "sms" | "push"
    status: str                      # "draft" | "active" | "archived"
    locales: list[TemplateLocale]    # multi-language content
    required_variables: list[str]    # enforced at render time
    version: int
    version_history: list[TemplateVersion]

Each TemplateLocale carries locale, subject, and body. required_variables is validated on every render — missing keys raise TemplateRenderError.

Registering custom templates

Custom templates are registered at startup before the lifespan seeds run:

from craft_easy.core.notifications.templates import register_notification_templates

register_notification_templates([
    {
        "name": "booking_confirmation",
        "channel": "email",
        "required_variables": ["booking_id", "start_time", "spot"],
        "locales": [
            {
                "locale": "en",
                "subject": "Booking confirmed",
                "body": "Your spot {{ spot }} is reserved from {{ start_time }}.",
            },
            {
                "locale": "sv",
                "subject": "Bokning bekräftad",
                "body": "Din plats {{ spot }} är reserverad från {{ start_time }}.",
            },
        ],
    }
])

The default templates (otp_email, otp_sms, welcome_email) are seeded automatically on first startup.

Delivery logs

Every send() writes a NotificationLog document to the notification_logs collection. This is your audit trail and is indexed on (channel, status), recipient, and user_id:

Field Purpose
channel, provider, recipient What was sent, how, to whom
subject, body_preview First ~200 characters for forensic lookup
status, message_id, error Result
attempts Total attempts across retries
user_id, application Linkage to the caller
delivered_at, metadata Provider-specific delivery info

Building a custom provider

Anything implementing NotificationProvider can be registered:

from craft_easy.core.notifications.base import (
    NotificationProvider,
    NotificationChannel,
    NotificationResult,
)

class TwilioProvider(NotificationProvider):
    channel = NotificationChannel.SMS
    name = "twilio"

    async def send(self, to, subject, body, **kwargs) -> NotificationResult:
        # Call Twilio API
        ...
        return NotificationResult(
            success=True,
            provider=self.name,
            channel=self.channel,
            recipient=to,
            message_id="SMxxx",
            error=None,
            attempts=1,
            metadata={},
        )

    async def is_healthy(self) -> bool:
        ...

# Register during app startup
from craft_easy.core.notifications import get_notification_registry
get_notification_registry().register(TwilioProvider(account_sid="...", auth_token="..."))

When multiple providers are registered for the same channel, the first one registered is the primary; the rest become ordered fallbacks.

Dry-run mode

Set NOTIFICATION_DRY_RUN=true to log every send without actually calling the provider. result.success is still True and a synthetic message_id is returned. This is the recommended default for CI and for local development without real credentials.

get_notification_registry().set_dry_run(True)