API Reference

Protocols

InboxStoreProtocol

source

Structural protocol for inbox message persistence backends.

All async methods are I/O-bound; implementations must not block the event loop. Callers may use either InMemoryInboxStore (for tests or simple deployments) or DatabaseInboxStore (for production SQL-backed storage).

save

async def save(message: InboxMessage) -> None

source

Persist message.

Parameters

Parameter	Type	Description
`message`	InboxMessage	The inbox message to store.

get

async def get(message_id: str) -> InboxMessage | None

source

Return the message with message_id, or None if not found.

Parameters

Parameter	Type	Description
`message_id`	str	The unique message ID to look up.

list_for_user

async def list_for_user(
    user_id: str,
    *,
    unread_only: bool = False
) -> list[InboxMessage]

source

Return messages for user_id in reverse-chronological order.

Parameters

Parameter	Type	Description
`user_id`	str	Filter to this user.
`unread_only`	bool	When ``True`` only unread messages are returned.

mark_read

async def mark_read(
    message_id: str,
    user_id: str
) -> None

source

Mark message_id as read, guarded by user_id ownership.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to mark read.
`user_id`	str	Owner of the message (for authorisation).

mark_all_read

async def mark_all_read(user_id: str) -> None

source

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

delete

async def delete(
    message_id: str,
    user_id: str
) -> None

source

Delete message_id, guarded by user_id ownership.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to remove.
`user_id`	str	Owner of the message (for authorisation).

count_unread

async def count_unread(user_id: str) -> int

source

Return the number of unread messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Unread message count.

clear_all

async def clear_all(user_id: str) -> int

source

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return backend health for inbox storage.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for the health probe.

---

MailerProtocol

source

Structural protocol for email delivery backends.

All mailer backends (SMTP, SendGrid, SES, etc.) must satisfy this protocol for Named DI resolution. send() accepts a fully-formed EmailMessage and returns Ok(receipt) on success or Err(MailerError) for expected delivery failures. Infrastructure failures (authentication, network timeout) must be raised as exceptions, not wrapped in Err.

send

async def send(message: EmailMessage) -> Result[MessageDeliveryReceipt, MailerError]

source

Send an email message.

Parameters

Parameter	Type	Description
`message`	EmailMessage	The email to deliver.

Returns

Type	Description
Result[MessageDeliveryReceipt, MailerError]	``Ok(MessageDeliveryReceipt)`` on acceptance by the backend. ``Err(MailerError)`` for expected delivery failures.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check backend connectivity.

Parameters

Parameter	Type	Description
`timeout`	float	Maximum seconds to wait for a response.

Returns

Type	Description
HealthCheckResult	HealthCheckResult with status details.

---

PushChannelProtocol

source

Structural protocol for push notification backends (FCM, APNS, etc.).

send() delivers to one or more device tokens. send_batch() delivers to multiple tokens in one API call where the provider supports it.

send

async def send(message: PushMessage) -> Result[MessageDeliveryReceipt, NotificationError]

source

Send a push notification to one or more device tokens.

Parameters

Parameter	Type	Description
`message`	PushMessage	The push notification to deliver.

Returns

Type	Description
Result[MessageDeliveryReceipt, NotificationError]	``Ok(MessageDeliveryReceipt)`` or ``Err(NotificationError)``.

send_batch

async def send_batch(messages: list[PushMessage]) -> list[Result[MessageDeliveryReceipt, NotificationError]]

source

Send push notifications in bulk.

Parameters

Parameter	Type	Description
`messages`	list[PushMessage]	List of push notifications to deliver.

Returns

Type	Description
list[Result[MessageDeliveryReceipt, NotificationError]]	List of ``Result`` values, one per message, preserving order.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check push backend connectivity.

---

SMSChannelProtocol

source

Structural protocol for SMS backends (Twilio, Vonage, etc.).

send() returns Ok(receipt) on acceptance or Err(NotificationError) for expected failures. Infrastructure failures must be raised as exceptions.

send

async def send(message: SMSMessage) -> Result[MessageDeliveryReceipt, NotificationError]

source

Send an SMS message.

Parameters

Parameter	Type	Description
`message`	SMSMessage	The SMS to deliver.

Returns

Type	Description
Result[MessageDeliveryReceipt, NotificationError]	``Ok(MessageDeliveryReceipt)`` or ``Err(NotificationError)``.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check SMS backend connectivity.

---

Classes

APNsDriverConfig

source

Apple Push Notification service (APNs) configuration.

---

DatabaseInboxStore

source

SQL-backed InboxStoreProtocol implementation.

Reads and writes to a single flat table. The table must exist before the store is used; the expected DDL is

CREATE TABLE notification_inbox_messages (
    id          TEXT        PRIMARY KEY,
    user_id     TEXT        NOT NULL,
    title       TEXT        NOT NULL,
    body        TEXT        NOT NULL,
    read        BOOLEAN     NOT NULL DEFAULT FALSE,
    created_at  TIMESTAMPTZ NOT NULL,
    metadata    JSONB       NOT NULL DEFAULT '{}'
);
CREATE INDEX ON notification_inbox_messages (user_id, created_at DESC);

CREATE TABLE notification_inbox_messages (
    id          TEXT        PRIMARY KEY,
    user_id     TEXT        NOT NULL,
    title       TEXT        NOT NULL,
    body        TEXT        NOT NULL,
    read        BOOLEAN     NOT NULL DEFAULT FALSE,
    created_at  TIMESTAMPTZ NOT NULL,
    metadata    JSONB       NOT NULL DEFAULT '{}'
);
CREATE INDEX ON notification_inbox_messages (user_id, created_at DESC);

Parameters

Parameter	Type	Description
`db`		Database provider injected via DI.
`table`		Table name. Defaults to ``notification_inbox_messages``.

__init__

def __init__(
    db: DatabaseProviderProtocol,
    *,
    table: str = _DEFAULT_TABLE
) -> None

source

save

async def save(message: InboxMessage) -> None

source

Persist message in the database.

Parameters

Parameter	Type	Description
`message`	InboxMessage	The inbox message to store.

get

async def get(message_id: str) -> InboxMessage | None

source

Return the message with message_id, or None.

Parameters

Parameter	Type	Description
`message_id`	str	ID to look up.

list_for_user

async def list_for_user(
    user_id: str,
    *,
    unread_only: bool = False
) -> list[InboxMessage]

source

Return messages for user_id in reverse-chronological order.

Parameters

Parameter	Type	Description
`user_id`	str	Filter to this user.
`unread_only`	bool	When ``True`` only unread messages are returned.

mark_read

async def mark_read(
    message_id: str,
    user_id: str
) -> None

source

Mark message_id as read, guarded by user_id ownership.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to mark.
`user_id`	str	Expected owner.

mark_all_read

async def mark_all_read(user_id: str) -> None

source

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

delete

async def delete(
    message_id: str,
    user_id: str
) -> None

source

Delete message_id, guarded by user_id ownership.

Parameters

Parameter	Type	Description
`message_id`	str	ID to remove.
`user_id`	str	Expected owner.

count_unread

async def count_unread(user_id: str) -> int

source

Return unread message count for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

clear_all

async def clear_all(user_id: str) -> int

source

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Run a lightweight read-only database probe for inbox storage.

---

EmailBouncedEvent

source

Emitted when an email bounces (hard or soft).

Consumed by: bounce management, list hygiene, audit.

---

EmailSentEvent

source

Emitted when an email is successfully dispatched to the provider.

Consumed by: audit, analytics, delivery tracking.

---

FCMDriverConfig

source

Firebase Cloud Messaging (FCM) push notification configuration.

---

InMemoryInboxStore

source

In-process InboxStoreProtocol implementation backed by a dict.

Thread-safety note: this store is not thread-safe. Within a single async event loop it is safe; for multi-threaded scenarios inject a SQL-backed DatabaseInboxStore instead.

__init__

def __init__() -> None

source

save

async def save(message: InboxMessage) -> None

source

Persist message in memory.

Parameters

Parameter	Type	Description
`message`	InboxMessage	The inbox message to store.

get

async def get(message_id: str) -> InboxMessage | None

source

Return the message with message_id, or None.

Parameters

Parameter	Type	Description
`message_id`	str	ID to look up.

list_for_user

async def list_for_user(
    user_id: str,
    *,
    unread_only: bool = False
) -> list[InboxMessage]

source

Return messages for user_id in reverse-chronological order.

Parameters

Parameter	Type	Description
`user_id`	str	Filter to this user.
`unread_only`	bool	When ``True`` only unread messages are returned.

mark_read

async def mark_read(
    message_id: str,
    user_id: str
) -> None

source

Mark message_id as read if it belongs to user_id.

InboxMessage is frozen; we create a replacement with read=True and update the store in-place.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to mark.
`user_id`	str	Expected owner.

mark_all_read

async def mark_all_read(user_id: str) -> None

source

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

delete

async def delete(
    message_id: str,
    user_id: str
) -> None

source

Delete message_id if it belongs to user_id.

Parameters

Parameter	Type	Description
`message_id`	str	ID to remove.
`user_id`	str	Expected owner.

count_unread

async def count_unread(user_id: str) -> int

source

Return unread message count for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

clear_all

async def clear_all(user_id: str) -> int

source

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Return health for the in-memory inbox backend.

---

InboxConfig

source

Top-level inbox configuration.

Loaded from the inbox: key in application.yaml, with environment variable overrides via LEX_NOTIFICATION__INBOX__* prefix.

Attributes: store_backend: Storage backend to use. One of 'database' or 'memory'. Defaults to 'database'. max_page_size: Maximum number of messages returned per page. retention_days: Number of days to retain inbox messages before they are eligible for pruning. mark_read_on_fetch: When True messages are automatically marked as read when fetched via get_inbox.

---

InboxMessage

source

A persisted inbox notification for a single user.

Attributes: id: Unique message identifier (UUID4 string). user_id: Recipient user ID. title: Short display title shown in notification lists. body: Full notification body text. read: Whether the recipient has read this message. created_at: UTC timestamp of creation. metadata: Opaque key-value payload for application-level use.

create

def create(
    cls,
    user_id: str,
    title: str,
    body: str,
    *,
    metadata: dict[str, Any] | None = None
) -> InboxMessage

source

Factory that auto-generates id and created_at.

Parameters

Parameter	Type	Description
`user_id`	str	Recipient user ID.
`title`	str	Short notification title.
`body`	str	Full notification body.
`metadata`	dict[str, Any] | None	Optional key-value payload.

Returns

Type	Description
InboxMessage	A new unsaved InboxMessage instance.

---

InboxMessageCreatedEvent

source

Emitted when a new message is added to a user's inbox.

Consumed by: audit, analytics, delivery tracking.

---

InboxMessageReadEvent

source

Emitted when a user reads an inbox message.

Consumed by: audit, analytics, engagement tracking.

---

InboxService

source

High-level service for managing per-user inbox notifications.

Wraps an InboxStoreProtocol backend and exposes a clean API consumed by controllers, event handlers, or a notification bell widget.

Defaults to InMemoryInboxStore when no store is supplied so that the service can be used without DI wiring in simple scenarios (e.g. tests, CLI tools). Production deployments should inject a DatabaseInboxStore via the DI container.

Parameters

Parameter	Type	Description
`store`		Persistence backend. Defaults to ``InMemoryInboxStore()``.

__init__

def __init__(store: InboxStoreProtocol | None = None) -> None

source

send

async def send(
    user_id: str,
    title: str,
    body: str,
    **metadata: Any
) -> InboxMessage

source

Create and persist an inbox notification for user_id.

Keyword arguments beyond body are collected into the message metadata dict so callers can attach arbitrary context

await inbox.send(
    user_id="u1",
    title="Order shipped",
    body="Your order #123 is on the way.",
    order_id="123",
    tracking_url="https://example.com/track/123",
)

await inbox.send(
    user_id="u1",
    title="Order shipped",
    body="Your order #123 is on the way.",
    order_id="123",
    tracking_url="https://example.com/track/123",
)

Parameters

Parameter	Type	Description
`user_id`	str	Recipient user ID.
`title`	str	Short notification title shown in list views.
`body`	str	Full message body. **metadata: Arbitrary key-value pairs stored in ``metadata``.

Returns

Type	Description
InboxMessage	The newly created and persisted InboxMessage.

get_inbox

async def get_inbox(
    user_id: str,
    *,
    unread_only: bool = False
) -> list[InboxMessage]

source

Return inbox messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	User whose inbox to fetch.
`unread_only`	bool	When ``True`` only unread messages are returned.

Returns

Type	Description
list[InboxMessage]	Messages in reverse-chronological order.

get_message

async def get_message(message_id: str) -> InboxMessage | None

source

Return a single inbox message by ID.

Parameters

Parameter	Type	Description
`message_id`	str	Message to retrieve.

Returns

Type	Description
InboxMessage | None	The message, or ``None`` if not found.

mark_read

async def mark_read(
    message_id: str,
    user_id: str
) -> None

source

Mark message_id as read.

The user_id guard ensures a user can only mark their own messages.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to mark.
`user_id`	str	Owning user (authorisation guard).

mark_all_read

async def mark_all_read(user_id: str) -> None

source

Mark all of user_id’s messages as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

delete

async def delete(
    message_id: str,
    user_id: str
) -> None

source

Delete a message owned by user_id.

Parameters

Parameter	Type	Description
`message_id`	str	ID of the message to remove.
`user_id`	str	Owning user (authorisation guard).

count_unread

async def count_unread(user_id: str) -> int

source

Return the unread message count for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of unread messages.

clear_all

async def clear_all(user_id: str) -> int

source

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

---

Mailable

source

Base class for structured email templates.

Subclass Mailable to encapsulate the data and rendering logic for a specific email type. Call to_message to build the EmailMessage that is passed to send.

Example

class WelcomeMail(Mailable):
    def __init__(self, user_name: str, user_email: str) -> None:
        self.user_name = user_name
        self.user_email = user_email

    def to_message(self) -> EmailMessage:
        return EmailMessage(
            to=[self.user_email],
            subject=f"Welcome, {self.user_name}!",
            body=f"Hi {self.user_name}, welcome aboard.",
            html_body=f"<p>Hi <b>{self.user_name}</b>, welcome aboard.</p>",
        )

class WelcomeMail(Mailable):
    def __init__(self, user_name: str, user_email: str) -> None:
        self.user_name = user_name
        self.user_email = user_email

    def to_message(self) -> EmailMessage:
        return EmailMessage(
            to=[self.user_email],
            subject=f"Welcome, {self.user_name}!",
            body=f"Hi {self.user_name}, welcome aboard.",
            html_body=f"<p>Hi <b>{self.user_name}</b>, welcome aboard.</p>",
        )

to_message

def to_message() -> EmailMessage

source

Build the EmailMessage for this mailable.

Returns

Type	Description
EmailMessage	A fully populated EmailMessage.

---

MailerConfig

source

Top-level mailer configuration.

Loaded from the mailer: key in application.yaml, with environment variable overrides via LEX_NOTIFICATION__MAILER__* prefix.

from_named

def from_named(
    cls,
    entry: NamedMailerConfig
) -> MailerConfig

source

Build a single-backend MailerConfig from a NamedMailerConfig entry.

Used internally by MailerProvider to create per-backend configs from a multi-backend declaration.

Parameters

Parameter	Type	Description
`entry`	NamedMailerConfig	The named backend entry to materialise.

Returns

Type	Description
MailerConfig	A MailerConfig configured for the single named backend.

---

NamedMailerConfig

source

Configuration for a single named mailer backend.

Used in MailerConfig.backends to declare multiple mailer backends that the framework registers as named DI bindings.

Example

backends:

• name: transactional driver: sendgrid primary: true sendgrid: api_key: sg_…
• name: internal driver: smtp smtp: host: smtp.example.com port: 587

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed MailerProtocol binding.
`driver`		Mailer driver. One of 'smtp' or 'sendgrid'.
`from_email`		Default sender email address.
`from_name`		Default sender display name.
`smtp`		SMTP-specific connection config.
`sendgrid`		SendGrid-specific config.

---

NamedPushConfig

source

Configuration for a single named push notification backend.

Used in NotificationConfig.push_backends to declare multiple push backends that the framework registers as named DI bindings.

Example

push_backends:

• name: mobile driver: fcm primary: true fcm: server_key: AAAA…

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed PushChannelProtocol binding.
`driver`		Push driver. One of 'fcm' or other supported drivers.
`fcm`		FCM-specific config.

---

NamedSMSConfig

source

Configuration for a single named SMS backend.

Used in NotificationConfig.sms_backends to declare multiple SMS backends that the framework registers as named DI bindings.

Example

sms_backends:

• name: alerts driver: twilio primary: true twilio: account_sid: AC… auth_token: … from_number: +1234567890

Parameters

Parameter	Type	Description
`name`		Unique backend identifier. Used as the Named() DI key.
`primary`		Whether this is the primary backend. Primary backends also receive the unnamed SMSChannelProtocol binding.
`driver`		SMS driver. One of 'twilio' or other supported drivers.
`twilio`		Twilio-specific config.

---

NotificationConfig

source

Top-level notification configuration.

Loaded from the notification: key in application.yaml, with environment variable overrides via LEX_NOTIFICATION__* prefix.

from_named_sms

def from_named_sms(
    cls,
    entry: NamedSMSConfig
) -> NotificationConfig

source

Build a single-SMS-backend NotificationConfig from a NamedSMSConfig entry.

Used internally by NotificationProvider to create per-backend configs from a multi-backend declaration.

Parameters

Parameter	Type	Description
`entry`	NamedSMSConfig	The named SMS backend entry to materialise.

Returns

Type	Description
NotificationConfig	A NotificationConfig configured for the single named SMS backend.

from_named_push

def from_named_push(
    cls,
    entry: NamedPushConfig
) -> NotificationConfig

source

Build a single-push-backend NotificationConfig from a NamedPushConfig entry.

Used internally by NotificationProvider to create per-backend configs from a multi-backend declaration.

Parameters

Parameter	Type	Description
`entry`	NamedPushConfig	The named push backend entry to materialise.

Returns

Type	Description
NotificationConfig	A NotificationConfig configured for the single named push backend.

---

NotificationFailedEvent

source

Emitted when a notification dispatch fails permanently.

Consumed by: audit, alerting, retry management.

---

NotificationFailedHook

source

Payload fired when a notification dispatch attempt fails.

Attributes: channel: Delivery channel that failed. recipient_id: Identifier of the intended recipient. reason: Short description of the failure.

---

NotificationModule

source

SMS and push notification integration with Named DI multi-backend support.

Registers SMSChannelProtocol and PushChannelProtocol for constructor injection.

Usage

from lexigram.notification.config import NotificationConfig
from lexigram.notification.module import NotificationModule

@module(
    imports=[NotificationModule.configure(NotificationConfig(backends=[...]))]
)
class AppModule(Module):
    pass

from lexigram.notification.config import NotificationConfig
from lexigram.notification.module import NotificationModule

@module(
    imports=[NotificationModule.configure(NotificationConfig(backends=[...]))]
)
class AppModule(Module):
    pass

Named injection

class MyService:
    def __init__(
        self,
        sms: SMSChannelProtocol,                                    # primary
        alerts_sms: Annotated[SMSChannelProtocol, Named("alerts")], # named
        push: PushChannelProtocol,                                   # primary
        mobile_push: Annotated[PushChannelProtocol, Named("mobile")], # named
    ) -> None: ...

class MyService:
    def __init__(
        self,
        sms: SMSChannelProtocol,                                    # primary
        alerts_sms: Annotated[SMSChannelProtocol, Named("alerts")], # named
        push: PushChannelProtocol,                                   # primary
        mobile_push: Annotated[PushChannelProtocol, Named("mobile")], # named
    ) -> None: ...

configure

def configure(
    cls,
    config: NotificationConfig | Any | None = None
) -> DynamicModule

source

Create a NotificationModule with explicit configuration.

Parameters

Parameter	Type	Description
`config`	NotificationConfig | Any | None	NotificationConfig or ``None`` to use defaults (reads from environment variables).

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return a NotificationModule for unit tests — no messages sent.

Uses an empty config (no backends configured) so all notifications are dropped and no external services are contacted.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

NotificationSentEvent

source

Emitted when a notification is successfully dispatched.

Consumed by: audit, analytics, delivery tracking.

---

NotificationSentHook

source

Payload fired when a notification is successfully dispatched.

Attributes: channel: Delivery channel used (e.g. "email", "sms", "push"). recipient_id: Identifier of the notification recipient.

---

PushMessage

source

A push notification message.

Attributes: to: List of device tokens (FCM registration IDs, APNS tokens, etc.). title: Notification title. body: Notification body text. data: Custom key-value payload delivered to the app. badge: iOS badge count; None leaves badge unchanged. sound: Notification sound name; None uses OS default. image: URL of a large notification image. ttl: Time-to-live in seconds; None uses provider default.

---

RetryingMailer

source

A decorator that wraps any MailerProtocol with exponential back-off retry and persistent delivery tracking.

Each send is assigned a stable delivery_id (UUID4). Every attempt is recorded via DeliveryStoreProtocol so that delivery history is preserved for auditing and dead-letter inspection.

Retry behaviour:

• Retries only on raised exceptions (infrastructure failures: SMTP timeouts, network errors, auth failures). These are the recoverable transient failures the retry pattern targets.
• Err(MailerError) returns from the inner mailer are not retried — they represent expected terminal delivery failures (bounced, rejected) per the MailerProtocol contract.
• After exhausting all attempts the final failure is logged, recorded as mark_failed(final=True), and returned as Err(MailerError) — never raised — to prevent cascading failures in notification flows.

Parameters

Parameter	Type	Description
`inner`		The underlying MailerProtocol implementation to delegate to.
`delivery_store`		Persistent store for delivery attempt tracking.
`max_attempts`		Maximum number of send attempts. Defaults to ``3``.
`base_delay`		Seconds to wait before the first retry. Each subsequent retry doubles the delay (exponential back-off). Defaults to ``1.0``.

__init__

def __init__(
    inner: MailerProtocol,
    delivery_store: DeliveryStoreProtocol,
    max_attempts: int = 3,
    base_delay: float = 1.0
) -> None

source

send

async def send(message: EmailMessage) -> Result[MessageDeliveryReceipt, MailerError]

source

Send an email with exponential back-off retry.

Conforms to MailerProtocol.

Parameters

Parameter	Type	Description
`message`	EmailMessage	The fully-formed email message to deliver.

Returns

Type	Description
Result[MessageDeliveryReceipt, MailerError]	``Ok(MessageDeliveryReceipt)`` on acceptance by the backend. ``Err(MailerError)`` when all attempts are exhausted or the inner mailer returns a terminal delivery failure.

---

SMSMessage

source

An SMS notification message.

Attributes: to: List of E.164-format recipient phone numbers. body: SMS body text (max 1600 chars; providers split if needed). from_number: Sender ID or phone number; backend default when None. metadata: Opaque provider-specific metadata.

---

SMTPDriverConfig

source

SMTP-specific connection configuration.

---

SMTPMailer

source

Email backend that sends via SMTP.

Implements MailerProtocol. SMTP operations are blocking; they are run in a thread pool via run_in_executor to avoid blocking the event loop.

Parameters

Parameter	Type	Description
`host`		SMTP server hostname.
`port`		SMTP port (587 for TLS, 465 for SSL, 25 for plain).
`username`		SMTP authentication username.
`password`		SMTP authentication password.
`use_tls`		Use STARTTLS after connection (port 587).
`use_ssl`		Use SSL from the start (port 465).
`timeout`		Connection timeout in seconds.
`from_email`		Default sender address.
`from_name`		Default sender display name.

__init__

def __init__(
    host: str = 'localhost',
    port: int = 587,
    username: str | None = None,
    password: str | None = None,
    use_tls: bool = True,
    use_ssl: bool = False,
    timeout: int = 30,
    from_email: str | None = None,
    from_name: str | None = None
) -> None

source

send

async def send(message: EmailMessage) -> Result[MessageDeliveryReceipt, MailerError]

source

Send an email via SMTP.

Parameters

Parameter	Type	Description
`message`	EmailMessage	The email message to send.

Returns

Type	Description
Result[MessageDeliveryReceipt, MailerError]	``Ok(MessageDeliveryReceipt)`` on success. ``Err(SMTPMailerError)`` for SMTP delivery failures.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check SMTP connectivity by attempting a connection.

Parameters

Parameter	Type	Description
`timeout`	float	Max seconds to wait for the connection.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

---

SendGridMailer

source

Email backend that sends via SendGrid REST API v3.

Implements MailerProtocol. Requires the aiohttp optional extra.

Parameters

Parameter	Type	Description
`api_key`		SendGrid API key (``SG.*``).
`timeout`		HTTP request timeout in seconds.
`sandbox_mode`		When ``True``, emails are not actually delivered.
`from_email`		Default sender email address.
`from_name`		Default sender display name.

__init__

def __init__(
    api_key: str,
    timeout: int = 30,
    sandbox_mode: bool = False,
    from_email: str | None = None,
    from_name: str | None = None
) -> None

source

send

async def send(message: EmailMessage) -> Result[MessageDeliveryReceipt, MailerError]

source

Send an email via the SendGrid REST API.

Parameters

Parameter	Type	Description
`message`	EmailMessage	The email to deliver.

Returns

Type	Description
Result[MessageDeliveryReceipt, MailerError]	``Ok(MessageDeliveryReceipt)`` on HTTP 202. ``Err(SendGridMailerError)`` for 4xx/5xx responses.

Raises

Exception	Description
aiohttp.ClientError	For network connectivity issues (DNS, connection refused).
OSError	For low-level socket/infrastructure errors.

Note

Infrastructure errors propagate to enable retry and circuit-breaker logic at higher levels. Only HTTP-level errors are wrapped in the Result type.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check SendGrid API reachability.

Parameters

Parameter	Type	Description
`timeout`	float	Max seconds to wait.

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

---

TwilioDriverConfig

source

Twilio SMS delivery configuration.

---

WebPushChannel

source

Web Push (RFC 8030) notification backend.

Implements PushChannelProtocol. Uses VAPID (RFC 8292) for application-level authentication and RFC 8291 (aes128gcm) for message encryption.

Requires the pywebpush optional extra.

Parameters

Parameter	Type	Description
`vapid_private_key`		VAPID private key (PEM-encoded EC prime256v1).
`vapid_public_key`		VAPID public key (base64url-encoded).
`vapid_claims_subject`		Contact URI for VAPID claims (e.g. ``"mailto:ops@example.com"``).
`http_timeout`		HTTP request timeout in seconds.

__init__

def __init__(
    vapid_private_key: str,
    vapid_public_key: str,
    vapid_claims_subject: str,
    http_timeout: int = _HTTP_CLIENT_TIMEOUT
) -> None

source

send

async def send(message: PushMessage) -> Result[MessageDeliveryReceipt, NotificationError]

source

Send a Web Push notification.

message.to contains a list of endpoint URLs (not device tokens). Each endpoint is a separate push subscription.

Parameters

Parameter	Type	Description
`message`	PushMessage	The push notification to deliver. ``message.to`` must contain at least one endpoint URL.

Returns

Type	Description
Result[MessageDeliveryReceipt, NotificationError]	Ok(MessageDeliveryReceipt) on success. Err(WebPushNotificationError) for delivery failures.

send_batch

async def send_batch(messages: list[PushMessage]) -> list[Result[MessageDeliveryReceipt, NotificationError]]

source

Send multiple Web Push notifications concurrently.

Parameters

Parameter	Type	Description
`messages`	list[PushMessage]	Notifications to deliver.

Returns

Type	Description
list[Result[MessageDeliveryReceipt, NotificationError]]	List of Result values, one per message, order preserved.

health_check

async def health_check(timeout: float = 5.0) -> HealthCheckResult

source

Check Web Push infrastructure availability.

Since Web Push has no global health endpoint, this is a best-effort check that verifies the VAPID key is loadable.

Parameters

Parameter	Type	Description
`timeout`	float	Max seconds to wait (unused, but kept for protocol compat).

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

---

WebPushDriverConfig

source

Web Push (RFC 8030) driver configuration.

---

Exceptions

APNsNotificationError

source

APNs push notification delivery failure.

__init__

def __init__(
    message: str = 'APNs push error',
    *,
    apns_reason: str | None = None,
    **kwargs: Any
) -> None

source

---

FCMNotificationError

source

FCM push notification delivery failure.

__init__

def __init__(
    message: str = 'FCM push error',
    *,
    fcm_error: str | None = None,
    **kwargs: Any
) -> None

source

---

InboxError

source

Base exception for all inbox submodule errors.

---

InboxMessageNotFoundError

source

Raised when a requested inbox message does not exist.

---

InboxPermissionError

source

Raised when a user attempts to access another user's inbox messages.

---

NotificationError

source

Base for expected, recoverable notification delivery failures.

Attributes: channel: Notification channel (e.g. “sms”, “push”). backend: Name of the backend (e.g. “twilio”, “fcm”).

__init__

def __init__(
    message: str = 'Notification error',
    *,
    channel: str = 'unknown',
    backend: str = 'unknown',
    **kwargs: Any
) -> None

source

---

SMTPMailerError

source

SMTP-specific delivery failure (rejection, auth error, etc.).

__init__

def __init__(
    message: str = 'SMTP delivery error',
    *,
    smtp_code: int | None = None,
    **kwargs: Any
) -> None

source

---

SendGridMailerError

source

SendGrid API delivery failure.

__init__

def __init__(
    message: str = 'SendGrid delivery error',
    *,
    status_code: int | None = None,
    **kwargs: Any
) -> None

source

---

TwilioNotificationError

source

Twilio SMS delivery failure.

__init__

def __init__(
    message: str = 'Twilio SMS error',
    *,
    twilio_code: int | None = None,
    **kwargs: Any
) -> None

source

---

WebPushNotificationError

source

Web Push notification delivery failure.

__init__

def __init__(
    message: str = 'Web Push error',
    *,
    status_code: int = 0,
    reason: str | None = None,
    **kwargs: Any
) -> None

source

---