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