Skip to content
PackageRequiredPurpose
quadkitYesCore framework
quadkit-contractsYesProtocol definitions
sendgridRecommendedEmail backend
twilioRecommendedSMS backend
firebase-adminOptionalFCM push backend

Your application needs to send notifications across multiple channels — email receipts, SMS alerts, push notifications to mobile devices, and in-app inbox messages. Each channel has different providers (Twilio, SendGrid, FCM, APNs), different protocols, and different configuration. Wiring them all manually leads to tightly coupled code and makes switching providers risky.

quadkit-notification solves this by defining channel protocols (SMSChannelProtocol, PushChannelProtocol, MailerProtocol) in quadkit-contracts, providing driver implementations for each major provider, and registering them in the DI container under Named bindings for multi-backend support.


┌──────────────────────────────┐
│ NotificationModule │
│ (SMS + Push) │
│ exports: │
│ SMSChannelProtocol │
│ PushChannelProtocol │
└───────────┬──────────────────┘
│
┌─────────────────────────────┼──────────────────────────┐
│ │ │
┌────▼─────┐ ┌───────▼────────┐ ┌───────▼────────┐
│ TwilioSMS │ │ FCMPush │ │ APNsPush │
│ (sms) │ │ WebPushChannel │ │ (ios) │
└──────────┘ └────────────────┘ └────────────────┘
▲
┌─────────────────┴──────────────────┐
│ MailerProvider (entry-point) │
│ exports: │
│ MailerProtocol │
└─────────────────┬──────────────────┘
│
┌───────────▼────────────┐
│ SendGridMailer │
│ SMTPMailer │
└────────────────────────┘

Three independent subsystems, all in one package:

SubsystemProviderProtocolsConfig Section
SMSNotificationProviderSMSChannelProtocolnotification.sms_backends
PushNotificationProviderPushChannelProtocolnotification.push_backends
EmailMailerProvider (auto-discovered)MailerProtocolmailer.backends
InboxStandalone servicesInboxStoreProtocolinbox.*

Every notification channel is abstracted behind a protocol from quadkit-contracts:

ProtocolImportsend() Signature
SMSChannelProtocolquadkit.contracts.notification.protocolssend(SMSMessage) -> Result[MessageDeliveryReceipt, NotificationError]
PushChannelProtocolquadkit.contracts.notification.protocolssend(PushMessage) -> Result[...] + send_batch(list[PushMessage])
MailerProtocolquadkit.contracts.mailer.protocolssend(EmailMessage) -> Result[MessageDeliveryReceipt, MailerError]

All send() methods return Result — expected failures (invalid number, quota exceeded) are Err, infrastructure errors (network timeout, auth failure) raise exceptions.

TypeFields
SMSMessageto: list[str], body: str, from_number, metadata
PushMessageto: list[str], title, body, data, badge, sound, image, ttl
InboxMessageid, user_id, title, body, read, created_at, metadata

Each channel supports multiple named backends. For example, two SMS providers:

notification:
sms_backends:
- name: primary
primary: true
driver: twilio
twilio:
account_sid: AC...
from_number: +15551111111
- name: fallback
driver: twilio
twilio:
account_sid: AC...
from_number: +15552222222

The primary backend gets an unnamed SMSChannelProtocol binding. All backends get Annotated[SMSChannelProtocol, Named(name)] for named injection:

from typing import Annotated
from quadkit.contracts.notification.protocols import SMSChannelProtocol
from quadkit.di.markers import Named
class AlertService:
def __init__(
self,
default_sms: SMSChannelProtocol, # primary
fallback_sms: Annotated[SMSChannelProtocol, Named("fallback")], # named
) -> None:
...

from quadkit.notification.module import NotificationModule
from quadkit.notification.config import NotificationConfig
from quadkit import Application
config = NotificationConfig(
sms_backends=[...],
push_backends=[...],
)
async with Application.boot(
name="my-app",
modules=[NotificationModule.configure(config)],
) as app:
...

The MailerProvider is auto-discovered via the quadkit.providers entry point. Configure it under mailer: in YAML:

application.yaml
mailer:
backends:
- name: transactional
primary: true
driver: sendgrid
from_email: [email protected]
sendgrid:
api_key: "${SENDGRID_API_KEY}"

Or configure programmatically and add the provider manually:

from quadkit.notification.di.mailer_provider import MailerProvider
from quadkit.notification.config import MailerConfig, NamedMailerConfig
provider = MailerProvider(config=MailerConfig(backends=[...]))
app.add_provider(provider)
from quadkit.contracts.mailer.types import EmailMessage
from quadkit.contracts.mailer.protocols import MailerProtocol
mailer = await container.resolve(MailerProtocol)
result = await mailer.send(
EmailMessage(
subject="Your Order Confirmation",
body=f"Order {order_id} confirmed.",
)
)
if result.is_err():
logger.error("email_failed", error=result.unwrap_err())

The inbox subsystem stores messages for in-app notification centers:

from quadkit.notification.inbox.service import InboxService
from quadkit.notification.inbox.memory import InMemoryInboxStore
# Ephemeral, single-process example; do not use for durable storage.
svc = InboxService(InMemoryInboxStore())
# Trusted application code chooses the recipient.
msg = await svc.send(user_id="user-42", title="Welcome!", body="Thanks for joining")
messages = await svc.get_inbox("user-42", limit=20, offset=0)
unread = await svc.count_unread("user-42")
own_message = await svc.get_message_for_user(msg.id, "user-42")

InboxModule.configure() defaults to the database backend and requires a registered DatabaseProviderProtocol; it fails when one is absent. For tests, use InboxModule.stub() or explicitly configure the memory backend. Calling InboxService() directly without a store also uses volatile memory.

These APIs do not authenticate or authorize callers. Request handlers must derive the user ID from verified server-side identity, scope every operation to that subject, and restrict send to trusted code or an explicitly authorized policy. get_message(id) is unscoped and is for trusted internal use; get_message_for_user(id, user_id) checks ownership but does not authenticate its caller. The retention_days setting does not itself schedule pruning.

quadkit.notification.inbox.http.InboxHttpAdapter supplies Starlette-compatible recipient inbox handlers for list, get, mark-read and delete. It requires an async subject resolver supplied by the host; use verified authentication state, not a user ID from request input. The host must also provide CSRF protection as appropriate. It does not expose a send endpoint. A cross-user lookup is 404; mark-read and delete return 204 for both absent and inaccessible IDs. The adapter currently omits message metadata from its response. The local SQL store performs owner-scoped, bounded page queries; legacy custom stores may load the full inbox before slicing. The service rejects list results with a mismatched owner or an oversized page from an optional paged store. These checks do not make a custom store trustworthy: mutations and unread counts still depend on its implementation. InboxModule does not mount the routes. Playground can mount the adapter only for its narrowly supported fresh local-source inbox composition; it does not migrate generic CRUD.


  • ✅ Use Result.match() or result.is_ok() to handle delivery outcomes — don’t unwrap blindly
  • ✅ Pin each backend with a descriptive name for Named() injection
  • ✅ Mark one backend per channel as primary: true for the unnamed default binding
  • ✅ Use NotificationModule.stub() in tests — it creates an empty config that drops all messages
  • ❌ Don’t store credentials in YAML — use SecretStr env-var overrides (QK_NOTIFICATION__*)
  • ❌ Don’t mix SMS and push credentials in the same backend config; use separate entries

  • Architecture — provider lifecycle, contracts, extension points
  • How-Tos — SMTP email, FCM push, named backends, inbox
  • Configuration — all config keys with defaults and env vars