API Reference

Protocols

`InboxStoreProtocol`

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

async def save(message: InboxMessage) -> None

Persist message.

Parameters

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

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

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.

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

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.

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

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

async def mark_all_read(user_id: str) -> None

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

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

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

async def count_unread(user_id: str) -> int

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.

async def clear_all(user_id: str) -> int

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

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

Return backend health for inbox storage.

Parameters

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

`MailerProtocol`

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.

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

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.

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

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`

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.

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

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

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

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.

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

Check push backend connectivity.

`SMSChannelProtocol`

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.

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

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

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

Check SMS backend connectivity.

Classes

`APNsDriverConfig`

Apple Push Notification service (APNs) configuration.

`DatabaseInboxStore`

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);

Parameters

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

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

async def save(message: InboxMessage) -> None

Persist message in the database.

Parameters

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

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

Return the message with message_id, or None.

Parameters

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

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

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.

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

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.

async def mark_all_read(user_id: str) -> None

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

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

Delete message_id, guarded by user_id ownership.

Parameters

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

async def count_unread(user_id: str) -> int

Return unread message count for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

async def clear_all(user_id: str) -> int

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

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

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

`EmailBouncedEvent`

Emitted when an email bounces (hard or soft).

Consumed by: bounce management, list hygiene, audit.

`EmailSentEvent`

Emitted when an email is successfully dispatched to the provider.

Consumed by: audit, analytics, delivery tracking.

`FCMDriverConfig`

Firebase Cloud Messaging (FCM) push notification configuration.

`InMemoryInboxStore`

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.

def __init__() -> None

async def save(message: InboxMessage) -> None

Persist message in memory.

Parameters

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

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

Return the message with message_id, or None.

Parameters

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

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

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.

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

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.

async def mark_all_read(user_id: str) -> None

Mark every unread message for user_id as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

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

Delete message_id if it belongs to user_id.

Parameters

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

async def count_unread(user_id: str) -> int

Return unread message count for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

async def clear_all(user_id: str) -> int

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

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

Return health for the in-memory inbox backend.

`InboxConfig`

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`

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.

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

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`

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

Consumed by: audit, analytics, delivery tracking.

`InboxMessageReadEvent`

Emitted when a user reads an inbox message.

Consumed by: audit, analytics, engagement tracking.

`InboxService`

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()``.

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

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

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",
)

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.

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

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.

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

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.

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

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

async def mark_all_read(user_id: str) -> None

Mark all of user_id’s messages as read.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

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

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

async def count_unread(user_id: str) -> int

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.

async def clear_all(user_id: str) -> int

Delete all messages for user_id.

Parameters

Parameter	Type	Description
`user_id`	str	Target user.

Returns

Type	Description
int	Number of messages deleted.

`Mailable`

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>",
        )

def to_message() -> EmailMessage

Build the EmailMessage for this mailable.

Returns

Type	Description
EmailMessage	A fully populated EmailMessage.

`MailerConfig`

Top-level mailer configuration.

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

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

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`

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`

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`

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`

Top-level notification configuration.

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

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

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.

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

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`

Emitted when a notification dispatch fails permanently.

Consumed by: audit, alerting, retry management.

`NotificationFailedHook`

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`

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

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

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

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.

def stub(cls) -> DynamicModule

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`

Emitted when a notification is successfully dispatched.

Consumed by: audit, analytics, delivery tracking.

`NotificationSentHook`

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`

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`

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

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

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

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`

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`

SMTP-specific connection configuration.

`SMTPMailer`

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.

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

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

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.

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

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`

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.

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

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

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.

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

Check SendGrid API reachability.

Parameters

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

Returns

Type	Description
HealthCheckResult	HealthCheckResult.

`TwilioDriverConfig`

Twilio SMS delivery configuration.

Exceptions

`APNsNotificationError`

APNs push notification delivery failure.

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

`FCMNotificationError`

FCM push notification delivery failure.

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

`InboxError`

Base exception for all inbox submodule errors.

`InboxMessageNotFoundError`

Raised when a requested inbox message does not exist.

`InboxPermissionError`

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

`NotificationError`

Base for expected, recoverable notification delivery failures.

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

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

`SMTPMailerError`

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

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

`SendGridMailerError`

SendGrid API delivery failure.

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

`TwilioNotificationError`

Twilio SMS delivery failure.

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