API Reference

Auto-generated from source docstrings. All public symbols are importable from the facades described below — you do not need to import from notificator.domain or notificator.infra directly.

Root package

The notificator root re-exports the domain primitives you need in every call site:

from notificator import (
    AsyncClosable,
    EmailAddress,
    NotificationClient,
    NotificationContent,
    NotificationError,
    PhoneNumber,
)

notificator.domain.value_objects

Domain value objects used by notification clients.

NotificationContent dataclass

A unified domain object representing the notification payload.

Source code in src/notificator/domain/value_objects.py

@dataclass(frozen=True, slots=True)
class NotificationContent:
    """A unified domain object representing the notification payload."""

    body: str
    subject: str | None = None
    html_body: str | None = None
    urgency: str | None = None

notificator.domain.types

Module containing static type markers used by notificator.

notificator.domain.ports

Domain-facing interfaces for notification delivery clients.

AsyncClosable

Bases: Protocol

Port for safely tearing down infrastructure adapters.

Source code in src/notificator/domain/ports.py

@runtime_checkable
class AsyncClosable(Protocol):
    """Port for safely tearing down infrastructure adapters."""

    async def aclose(self) -> None:
        """Release any held resources and close open connections."""

aclose() async

Release any held resources and close open connections.

Source code in src/notificator/domain/ports.py

async def aclose(self) -> None:
    """Release any held resources and close open connections."""

NotificationClient

Bases: Protocol

Protocol for notification transports (email, SMS, etc.).

Source code in src/notificator/domain/ports.py

class NotificationClient[RecipientT: str](Protocol):
    """Protocol for notification transports (email, SMS, etc.)."""

    async def notify(self, content: NotificationContent, *, recipient: RecipientT) -> None:
        """Send a notification payload to a recipient.

        Args:
            content: Notification payload describing body, subject, and other metadata.
            recipient: Transport-specific recipient identifier (email, phone, etc.).
        """

    async def notify_from_template(
        self, template: str, *, recipient: RecipientT, version: str | None = None, **variables: str
    ) -> None:
        """Send a notification using a stored template and variables.

        Args:
            template: Provider template identifier.
            recipient: Transport-specific recipient identifier (email, phone, etc.).
            version: Optional provider template version alias or ID.
            **variables: Template variables to inject into the message.
        """

notify(content, *, recipient) async

Send a notification payload to a recipient.

Parameters:

Name	Type	Description	Default
content	NotificationContent	Notification payload describing body, subject, and other metadata.	required
recipient	RecipientT	Transport-specific recipient identifier (email, phone, etc.).	required

Source code in src/notificator/domain/ports.py

async def notify(self, content: NotificationContent, *, recipient: RecipientT) -> None:
    """Send a notification payload to a recipient.

    Args:
        content: Notification payload describing body, subject, and other metadata.
        recipient: Transport-specific recipient identifier (email, phone, etc.).
    """

notify_from_template(template, *, recipient, version=None, **variables) async

Send a notification using a stored template and variables.

Parameters:

Name	Type	Description	Default
template	str	Provider template identifier.	required
recipient	RecipientT	Transport-specific recipient identifier (email, phone, etc.).	required
version	str | None	Optional provider template version alias or ID.	None
**variables	str	Template variables to inject into the message.	{}

Source code in src/notificator/domain/ports.py

async def notify_from_template(
    self, template: str, *, recipient: RecipientT, version: str | None = None, **variables: str
) -> None:
    """Send a notification using a stored template and variables.

    Args:
        template: Provider template identifier.
        recipient: Transport-specific recipient identifier (email, phone, etc.).
        version: Optional provider template version alias or ID.
        **variables: Template variables to inject into the message.
    """

notificator.domain.exceptions

Module for domain exceptions.

NotificationError

Bases: Exception

Base class for all notification-related exceptions.

Source code in src/notificator/domain/exceptions.py

class NotificationError(Exception):
    """Base class for all notification-related exceptions."""

Email — notificator.mail

from notificator.mail import MailgunClient

notificator.infra.mail_clients.mailgun_client

Mailgun-backed email notification client implementation.

MailgunClient

Bases: NotificationClient[EmailAddress], AsyncClosable

Send email notifications via the Mailgun HTTP API.

Source code in src/notificator/infra/mail_clients/mailgun_client.py

class MailgunClient(NotificationClient[EmailAddress], AsyncClosable):
    """Send email notifications via the Mailgun HTTP API."""

    __slots__ = (
        "_api_key",
        "_base_url",
        "_domain",
        "_http_client",
        "_sender_display_name",
        "_sender_email",
        "default_subject",
    )

    def __init__(  # noqa: PLR0913
        self,
        domain: str,
        *,
        default_subject: str | None = None,
        api_key: str | None = None,
        base_url: str = "https://api.eu.mailgun.net/v3",
        sender_email: str,
        sender_display_name: str,
        http_client: httpx.AsyncClient | None = None,
    ) -> None:
        """Initialize a Mailgun-backed notification client.

        Args:
            domain: Mailgun domain used to send messages.
            default_subject: Default subject line for emails without an explicit subject.
            api_key: Mailgun API key. Optional if `http_client` already has auth.
            base_url: Mailgun API base URL.
            sender_email: Email address used in the From header.
            sender_display_name: Display name used in the From header.
            http_client: Optional preconfigured `httpx.AsyncClient`.

        Raises:
            MissingClientAuthError: Raised when neither `api_key` nor authenticated
                `http_client` is provided.
            MalformedClientUrlError: Raised when `base_url` is not a valid URL.
        """
        if api_key is None and (http_client is None or http_client.auth is None):
            raise MissingClientAuthError

        try:
            normalized_url = _http_url_adapter.validate_python(base_url)
        except ValidationError as e:
            raise MalformedClientUrlError(base_url) from e

        self.default_subject = default_subject

        self._api_key = api_key
        self._base_url = str(normalized_url)
        self._domain = domain
        self._http_client = (
            http_client or httpx.AsyncClient(auth=("api", self._api_key))
            if self._api_key
            else httpx.AsyncClient()
        )
        self._sender_email = sender_email
        self._sender_display_name = sender_display_name

    async def aclose(self) -> None:
        """Explicitly close the HTTP client to prevent connection leaks."""
        await self._http_client.aclose()

    def _validate_email(self, recipient: EmailAddress) -> str:
        try:
            return _email_adapter.validate_python(recipient)
        except ValidationError as e:
            raise MalformedRecipientEmailError(recipient) from e

    async def _post(self, data: dict[str, str]) -> None:
        url = f"{self._base_url}/{self._domain}/messages"

        try:
            if self._http_client.auth is None:
                assert self._api_key is not None, (
                    "this should never happen and is likely a bug, please report it."
                )
                auth = ("api", self._api_key)
                response = await self._http_client.post(url, data=data, auth=auth)
            else:
                response = await self._http_client.post(url, data=data)
            response.raise_for_status()
        except httpx.HTTPError as e:
            raise MailAPIError from e

    async def notify_from_template(
        self,
        template: str,
        *,
        recipient: EmailAddress,
        version: str | None = None,
        **variables: str,
    ) -> None:
        """Send a Mailgun template-based email to a recipient.

        Args:
            template: Mailgun template name.
            recipient: Email address to receive the notification.
            version: Optional Mailgun template version.
            **variables: Template variables to interpolate.
        """
        valid_email = self._validate_email(recipient)

        data = {
            "from": f"{self._sender_display_name} <{self._sender_email}>",
            "to": valid_email,
            "template": template,
            "t:variables": json.dumps(variables),
        }
        if version:
            data["t:version"] = version

        await self._post(data)

    async def notify(self, content: NotificationContent, *, recipient: EmailAddress) -> None:
        """Send a plain email notification without templates.

        Args:
            content: Notification payload with body and optional subject.
            recipient: Email address to receive the notification.
        """
        if content.subject is None and self.default_subject is None:
            raise EmailNotificationMissingSubjectError

        valid_email = self._validate_email(recipient)

        data: dict[str, str] = {
            "from": f"{self._sender_display_name} <{self._sender_email}>",
            "to": valid_email,
            "subject": content.subject
            or self.default_subject
            or "New Notification",  # Only to satisfy mypy
            "text": content.body,
        }

        await self._post(data)

__init__(domain, *, default_subject=None, api_key=None, base_url='https://api.eu.mailgun.net/v3', sender_email, sender_display_name, http_client=None)

Initialize a Mailgun-backed notification client.

Parameters:

Name	Type	Description	Default
domain	str	Mailgun domain used to send messages.	required
default_subject	str | None	Default subject line for emails without an explicit subject.	None
api_key	str | None	Mailgun API key. Optional if http_client already has auth.	None
base_url	str	Mailgun API base URL.	'https://api.eu.mailgun.net/v3'
sender_email	str	Email address used in the From header.	required
sender_display_name	str	Display name used in the From header.	required
http_client	AsyncClient | None	Optional preconfigured httpx.AsyncClient.	None

Raises:

Type	Description
MissingClientAuthError	Raised when neither api_key nor authenticated http_client is provided.
MalformedClientUrlError	Raised when base_url is not a valid URL.

Source code in src/notificator/infra/mail_clients/mailgun_client.py

def __init__(  # noqa: PLR0913
    self,
    domain: str,
    *,
    default_subject: str | None = None,
    api_key: str | None = None,
    base_url: str = "https://api.eu.mailgun.net/v3",
    sender_email: str,
    sender_display_name: str,
    http_client: httpx.AsyncClient | None = None,
) -> None:
    """Initialize a Mailgun-backed notification client.

    Args:
        domain: Mailgun domain used to send messages.
        default_subject: Default subject line for emails without an explicit subject.
        api_key: Mailgun API key. Optional if `http_client` already has auth.
        base_url: Mailgun API base URL.
        sender_email: Email address used in the From header.
        sender_display_name: Display name used in the From header.
        http_client: Optional preconfigured `httpx.AsyncClient`.

    Raises:
        MissingClientAuthError: Raised when neither `api_key` nor authenticated
            `http_client` is provided.
        MalformedClientUrlError: Raised when `base_url` is not a valid URL.
    """
    if api_key is None and (http_client is None or http_client.auth is None):
        raise MissingClientAuthError

    try:
        normalized_url = _http_url_adapter.validate_python(base_url)
    except ValidationError as e:
        raise MalformedClientUrlError(base_url) from e

    self.default_subject = default_subject

    self._api_key = api_key
    self._base_url = str(normalized_url)
    self._domain = domain
    self._http_client = (
        http_client or httpx.AsyncClient(auth=("api", self._api_key))
        if self._api_key
        else httpx.AsyncClient()
    )
    self._sender_email = sender_email
    self._sender_display_name = sender_display_name

aclose() async

Explicitly close the HTTP client to prevent connection leaks.

Source code in src/notificator/infra/mail_clients/mailgun_client.py

async def aclose(self) -> None:
    """Explicitly close the HTTP client to prevent connection leaks."""
    await self._http_client.aclose()

notify(content, *, recipient) async

Send a plain email notification without templates.

Parameters:

Name	Type	Description	Default
content	NotificationContent	Notification payload with body and optional subject.	required
recipient	EmailAddress	Email address to receive the notification.	required

Source code in src/notificator/infra/mail_clients/mailgun_client.py

async def notify(self, content: NotificationContent, *, recipient: EmailAddress) -> None:
    """Send a plain email notification without templates.

    Args:
        content: Notification payload with body and optional subject.
        recipient: Email address to receive the notification.
    """
    if content.subject is None and self.default_subject is None:
        raise EmailNotificationMissingSubjectError

    valid_email = self._validate_email(recipient)

    data: dict[str, str] = {
        "from": f"{self._sender_display_name} <{self._sender_email}>",
        "to": valid_email,
        "subject": content.subject
        or self.default_subject
        or "New Notification",  # Only to satisfy mypy
        "text": content.body,
    }

    await self._post(data)

notify_from_template(template, *, recipient, version=None, **variables) async

Send a Mailgun template-based email to a recipient.

Parameters:

Name	Type	Description	Default
template	str	Mailgun template name.	required
recipient	EmailAddress	Email address to receive the notification.	required
version	str | None	Optional Mailgun template version.	None
**variables	str	Template variables to interpolate.	{}

Source code in src/notificator/infra/mail_clients/mailgun_client.py

async def notify_from_template(
    self,
    template: str,
    *,
    recipient: EmailAddress,
    version: str | None = None,
    **variables: str,
) -> None:
    """Send a Mailgun template-based email to a recipient.

    Args:
        template: Mailgun template name.
        recipient: Email address to receive the notification.
        version: Optional Mailgun template version.
        **variables: Template variables to interpolate.
    """
    valid_email = self._validate_email(recipient)

    data = {
        "from": f"{self._sender_display_name} <{self._sender_email}>",
        "to": valid_email,
        "template": template,
        "t:variables": json.dumps(variables),
    }
    if version:
        data["t:version"] = version

    await self._post(data)

notificator.infra.mail_clients.exceptions

Base module for mail clients exceptions.

EmailNotificationMissingSubjectError

Bases: MailNotificationError

Exception raised when an email notification is missing a subject.

Source code in src/notificator/infra/mail_clients/exceptions.py

class EmailNotificationMissingSubjectError(MailNotificationError):
    """Exception raised when an email notification is missing a subject."""

    def __init__(self) -> None:
        """Initialize the error for missing notification subject."""
        super().__init__(
            "You need to either provide a subject for email NotificationContent "
            "or set a default_subject for compatible clients."
        )

__init__()

Initialize the error for missing notification subject.

Source code in src/notificator/infra/mail_clients/exceptions.py

def __init__(self) -> None:
    """Initialize the error for missing notification subject."""
    super().__init__(
        "You need to either provide a subject for email NotificationContent "
        "or set a default_subject for compatible clients."
    )

MailAPIError

Bases: MailNotificationError

Exception raised when an error occurs while making an API call.

Source code in src/notificator/infra/mail_clients/exceptions.py

class MailAPIError(MailNotificationError):
    """Exception raised when an error occurs while making an API call."""

MailNotificationError

Bases: NotificationError

Base exception for mail clients errors.

Source code in src/notificator/infra/mail_clients/exceptions.py

class MailNotificationError(NotificationError):
    """Base exception for mail clients errors."""

MalformedClientUrlError

Bases: MailNotificationError

Exception raised when mail client base_url is malformed.

Source code in src/notificator/infra/mail_clients/exceptions.py

class MalformedClientUrlError(MailNotificationError):
    """Exception raised when mail client base_url is malformed."""

    def __init__(self, base_url: str) -> None:
        """Initialize the error with the invalid base URL.

        Args:
            base_url: The malformed base URL provided to the client.
        """
        self.base_url = base_url
        super().__init__(f"{self.base_url} is not a valid url.")

__init__(base_url)

Initialize the error with the invalid base URL.

Parameters:

Name	Type	Description	Default
base_url	str	The malformed base URL provided to the client.	required

Source code in src/notificator/infra/mail_clients/exceptions.py

def __init__(self, base_url: str) -> None:
    """Initialize the error with the invalid base URL.

    Args:
        base_url: The malformed base URL provided to the client.
    """
    self.base_url = base_url
    super().__init__(f"{self.base_url} is not a valid url.")

MalformedRecipientEmailError

Bases: MailNotificationError

Exception raised when a recipient email is malformed.

Source code in src/notificator/infra/mail_clients/exceptions.py

class MalformedRecipientEmailError(MailNotificationError):
    """Exception raised when a recipient email is malformed."""

    def __init__(self, recipient: str) -> None:
        """Initialize the error with the invalid recipient email.

        Args:
            recipient: The invalid email address provided to the client.
        """
        self.recipient = recipient
        super().__init__(f"{self.recipient} is not a valid email address")

__init__(recipient)

Initialize the error with the invalid recipient email.

Parameters:

Name	Type	Description	Default
recipient	str	The invalid email address provided to the client.	required

Source code in src/notificator/infra/mail_clients/exceptions.py

def __init__(self, recipient: str) -> None:
    """Initialize the error with the invalid recipient email.

    Args:
        recipient: The invalid email address provided to the client.
    """
    self.recipient = recipient
    super().__init__(f"{self.recipient} is not a valid email address")

MissingClientAuthError

Bases: MailNotificationError

Exception raised when a mailclient authorization is missing.

Source code in src/notificator/infra/mail_clients/exceptions.py

class MissingClientAuthError(MailNotificationError):
    """Exception raised when a mailclient authorization is missing."""

    def __init__(self) -> None:
        """Initialize the error for missing Mailgun authentication."""
        super().__init__("You need to either provide an `api_key` or a configured httpx client.")

__init__()

Initialize the error for missing Mailgun authentication.

Source code in src/notificator/infra/mail_clients/exceptions.py

def __init__(self) -> None:
    """Initialize the error for missing Mailgun authentication."""
    super().__init__("You need to either provide an `api_key` or a configured httpx client.")

SMS — notificator.sms

from notificator.sms import TwilioSmsClient, TwilioSmsTemplate

notificator.infra.sms_clients.twilio_sms_client

Twilio-backed SMS notification client implementation.

TwilioSmsClient

Bases: NotificationClient[PhoneNumber], AsyncClosable

Send SMS notifications via Twilio's REST API.

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

class TwilioSmsClient(NotificationClient[PhoneNumber], AsyncClosable):
    """Send SMS notifications via Twilio's REST API."""

    __slots__ = ("_client", "_messaging_service_sid", "_sender_phone_number", "_template_registry")

    def __init__(  # noqa: PLR0913
        self,
        account_sid: str,
        token: str,
        *,
        templates: list[TwilioSmsTemplate | str] | None = None,
        twilio_http_client: AsyncHttpClient | None = None,
        messaging_service_sid: str | None = None,
        sender_phone_number: PhoneNumber | None = None,
    ) -> None:
        """Initialize a Twilio-backed SMS notification client.

        Args:
            account_sid: Twilio account SID.
            token: Twilio auth token.
            templates: Optional list of templates, either str representing sid or
                TwilioSmsTemplate data objects, allowing custom versioning.
            twilio_http_client: Optional async HTTP client for Twilio.
            messaging_service_sid: Optional messaging service SID.
            sender_phone_number: Optional sender phone number in E.164 format.

        Raises:
            TwilioMissingSenderIdError: Raised when neither sender phone number nor
                messaging service SID is provided.
            InvalidPhoneNumberFormatError: Raised when `sender_phone_number` is invalid.
        """
        template_registry: dict[str, TwilioSmsTemplate] = {}

        if templates is not None:
            for t in templates:
                if isinstance(t, str):
                    template_registry[t] = TwilioSmsTemplate(id=t)
                else:
                    template_registry[t.id] = t

        if sender_phone_number is None and messaging_service_sid is None:
            raise TwilioMissingSenderIdError

        if sender_phone_number is not None:
            try:
                e164_sender_phone_number = _phone_number_adapter.validate_python(
                    sender_phone_number
                )
            except ValidationError as e:
                raise InvalidPhoneNumberFormatError(sender_phone_number) from e
        else:
            e164_sender_phone_number = None

        self._client = Client(
            account_sid, token, http_client=twilio_http_client or AsyncTwilioHttpClient()
        )
        self._messaging_service_sid = messaging_service_sid
        self._sender_phone_number = e164_sender_phone_number
        self._template_registry = template_registry

    def _validate_phone_number(self, phone_number: PhoneNumber) -> str:
        try:
            return _phone_number_adapter.validate_python(phone_number)
        except ValidationError as e:
            raise InvalidPhoneNumberFormatError(phone_number) from e

    async def aclose(self) -> None:
        """Close the underlying Twilio HTTP client if it is async-aware."""
        if isinstance(self._client.http_client, AsyncTwilioHttpClient):
            await self._client.http_client.close()

    async def notify(self, content: NotificationContent, *, recipient: PhoneNumber) -> None:
        """Send a plain SMS message to a recipient.

        Args:
            content: Notification payload containing the message body.
            recipient: E.164-compatible phone number to receive the SMS.
        """
        e164_recipient = self._validate_phone_number(recipient)

        try:
            await self._client.messages.create_async(
                from_=self._sender_phone_number or values.unset,
                messaging_service_sid=self._messaging_service_sid or values.unset,
                body=content.body,
                to=e164_recipient,
            )
        except TwilioException as e:
            raise SmsAPIError from e

    async def notify_from_template(
        self, template: str, *, recipient: PhoneNumber, version: str | None = None, **variables: str
    ) -> None:
        """Send a Twilio Content API template with injected variables.

        Args:
            template: Template identifier registered in Twilio.
            recipient: E.164-compatible phone number to receive the SMS.
            version: Optional template version alias or ID to resolve.
            **variables: Template variables to inject into the message.
        """
        e164_recipient = self._validate_phone_number(recipient)
        try:
            template_obj = self._template_registry[template]
        except KeyError as e:
            raise TemplateNotProvidedError(template) from e

        if version:
            try:
                version = template_obj.version_registry[version]
            except KeyError as e:
                raise TemplateVersionNotAvailableError(
                    template_name=template_obj.id, version=version
                ) from e
        try:
            variables_str = json.dumps(variables)
            await self._client.messages.create_async(
                from_=self._sender_phone_number or values.unset,
                messaging_service_sid=self._messaging_service_sid or values.unset,
                content_sid=template,
                content_variables=variables_str,
                to=e164_recipient,
            )
        except TwilioException as e:
            raise SmsAPIError from e

__init__(account_sid, token, *, templates=None, twilio_http_client=None, messaging_service_sid=None, sender_phone_number=None)

Initialize a Twilio-backed SMS notification client.

Parameters:

Name	Type	Description	Default
account_sid	str	Twilio account SID.	required
token	str	Twilio auth token.	required
templates	list[TwilioSmsTemplate | str] | None	Optional list of templates, either str representing sid or TwilioSmsTemplate data objects, allowing custom versioning.	None
twilio_http_client	AsyncHttpClient | None	Optional async HTTP client for Twilio.	None
messaging_service_sid	str | None	Optional messaging service SID.	None
sender_phone_number	PhoneNumber | None	Optional sender phone number in E.164 format.	None

Raises:

Type	Description
TwilioMissingSenderIdError	Raised when neither sender phone number nor messaging service SID is provided.
InvalidPhoneNumberFormatError	Raised when sender_phone_number is invalid.

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

def __init__(  # noqa: PLR0913
    self,
    account_sid: str,
    token: str,
    *,
    templates: list[TwilioSmsTemplate | str] | None = None,
    twilio_http_client: AsyncHttpClient | None = None,
    messaging_service_sid: str | None = None,
    sender_phone_number: PhoneNumber | None = None,
) -> None:
    """Initialize a Twilio-backed SMS notification client.

    Args:
        account_sid: Twilio account SID.
        token: Twilio auth token.
        templates: Optional list of templates, either str representing sid or
            TwilioSmsTemplate data objects, allowing custom versioning.
        twilio_http_client: Optional async HTTP client for Twilio.
        messaging_service_sid: Optional messaging service SID.
        sender_phone_number: Optional sender phone number in E.164 format.

    Raises:
        TwilioMissingSenderIdError: Raised when neither sender phone number nor
            messaging service SID is provided.
        InvalidPhoneNumberFormatError: Raised when `sender_phone_number` is invalid.
    """
    template_registry: dict[str, TwilioSmsTemplate] = {}

    if templates is not None:
        for t in templates:
            if isinstance(t, str):
                template_registry[t] = TwilioSmsTemplate(id=t)
            else:
                template_registry[t.id] = t

    if sender_phone_number is None and messaging_service_sid is None:
        raise TwilioMissingSenderIdError

    if sender_phone_number is not None:
        try:
            e164_sender_phone_number = _phone_number_adapter.validate_python(
                sender_phone_number
            )
        except ValidationError as e:
            raise InvalidPhoneNumberFormatError(sender_phone_number) from e
    else:
        e164_sender_phone_number = None

    self._client = Client(
        account_sid, token, http_client=twilio_http_client or AsyncTwilioHttpClient()
    )
    self._messaging_service_sid = messaging_service_sid
    self._sender_phone_number = e164_sender_phone_number
    self._template_registry = template_registry

aclose() async

Close the underlying Twilio HTTP client if it is async-aware.

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

async def aclose(self) -> None:
    """Close the underlying Twilio HTTP client if it is async-aware."""
    if isinstance(self._client.http_client, AsyncTwilioHttpClient):
        await self._client.http_client.close()

notify(content, *, recipient) async

Send a plain SMS message to a recipient.

Parameters:

Name	Type	Description	Default
content	NotificationContent	Notification payload containing the message body.	required
recipient	PhoneNumber	E.164-compatible phone number to receive the SMS.	required

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

async def notify(self, content: NotificationContent, *, recipient: PhoneNumber) -> None:
    """Send a plain SMS message to a recipient.

    Args:
        content: Notification payload containing the message body.
        recipient: E.164-compatible phone number to receive the SMS.
    """
    e164_recipient = self._validate_phone_number(recipient)

    try:
        await self._client.messages.create_async(
            from_=self._sender_phone_number or values.unset,
            messaging_service_sid=self._messaging_service_sid or values.unset,
            body=content.body,
            to=e164_recipient,
        )
    except TwilioException as e:
        raise SmsAPIError from e

notify_from_template(template, *, recipient, version=None, **variables) async

Send a Twilio Content API template with injected variables.

Parameters:

Name	Type	Description	Default
template	str	Template identifier registered in Twilio.	required
recipient	PhoneNumber	E.164-compatible phone number to receive the SMS.	required
version	str | None	Optional template version alias or ID to resolve.	None
**variables	str	Template variables to inject into the message.	{}

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

async def notify_from_template(
    self, template: str, *, recipient: PhoneNumber, version: str | None = None, **variables: str
) -> None:
    """Send a Twilio Content API template with injected variables.

    Args:
        template: Template identifier registered in Twilio.
        recipient: E.164-compatible phone number to receive the SMS.
        version: Optional template version alias or ID to resolve.
        **variables: Template variables to inject into the message.
    """
    e164_recipient = self._validate_phone_number(recipient)
    try:
        template_obj = self._template_registry[template]
    except KeyError as e:
        raise TemplateNotProvidedError(template) from e

    if version:
        try:
            version = template_obj.version_registry[version]
        except KeyError as e:
            raise TemplateVersionNotAvailableError(
                template_name=template_obj.id, version=version
            ) from e
    try:
        variables_str = json.dumps(variables)
        await self._client.messages.create_async(
            from_=self._sender_phone_number or values.unset,
            messaging_service_sid=self._messaging_service_sid or values.unset,
            content_sid=template,
            content_variables=variables_str,
            to=e164_recipient,
        )
    except TwilioException as e:
        raise SmsAPIError from e

TwilioSmsTemplate dataclass

Template metadata for Twilio Content API usage.

Attributes:

Name	Type	Description
id	str	Twilio Content SID for the template or a friendly name if using registry anyway.
version_registry	dict[str, str]	Optional mapping of friendly version names to Content SIDs.

Source code in src/notificator/infra/sms_clients/twilio_sms_client.py

@dataclass(frozen=True, slots=True)
class TwilioSmsTemplate:
    """Template metadata for Twilio Content API usage.

    Attributes:
        id: Twilio Content SID for the template or a friendly name if using registry anyway.
        version_registry: Optional mapping of friendly version names to Content SIDs.
    """

    id: str
    version_registry: dict[str, str] = field(default_factory=dict)

notificator.infra.sms_clients.exceptions

Base module for sms clients exceptions.

InvalidPhoneNumberFormatError

Bases: SmsNotificationError

Raised when a phone number format is invalid for a client.

Source code in src/notificator/infra/sms_clients/exceptions.py

class InvalidPhoneNumberFormatError(SmsNotificationError):
    """Raised when a phone number format is invalid for a client."""

    def __init__(self, phone_number: str, expected_format: str = "E164") -> None:
        """Initialize the error with the invalid phone number and expected format.

        Args:
            phone_number: The provided phone number.
            expected_format: The expected phone number format.
        """
        self.phone_number = phone_number
        self.expected_format = expected_format
        super().__init__(
            f"Provided phone number {self.phone_number} does not match expected format: {self.expected_format}"
        )

__init__(phone_number, expected_format='E164')

Initialize the error with the invalid phone number and expected format.

Parameters:

Name	Type	Description	Default
phone_number	str	The provided phone number.	required
expected_format	str	The expected phone number format.	'E164'

Source code in src/notificator/infra/sms_clients/exceptions.py

def __init__(self, phone_number: str, expected_format: str = "E164") -> None:
    """Initialize the error with the invalid phone number and expected format.

    Args:
        phone_number: The provided phone number.
        expected_format: The expected phone number format.
    """
    self.phone_number = phone_number
    self.expected_format = expected_format
    super().__init__(
        f"Provided phone number {self.phone_number} does not match expected format: {self.expected_format}"
    )

SmsAPIError

Bases: SmsNotificationError

Exception raised when an error occurs while making an API call.

Source code in src/notificator/infra/sms_clients/exceptions.py

class SmsAPIError(SmsNotificationError):
    """Exception raised when an error occurs while making an API call."""

SmsNotificationError

Bases: NotificationError

Base exception for sms notification errors.

Source code in src/notificator/infra/sms_clients/exceptions.py

class SmsNotificationError(NotificationError):
    """Base exception for sms notification errors."""

TemplateNotProvidedError

Bases: SmsNotificationError

Raised when requested to use a template that is not available for this client.

Source code in src/notificator/infra/sms_clients/exceptions.py

class TemplateNotProvidedError(SmsNotificationError):
    """Raised when requested to use a template that is not available for this client."""

    def __init__(self, template_name: str) -> None:
        """Initialize the error with the missing template name.

        Args:
            template_name: The template identifier not configured for this client.
        """
        self.template_name = template_name
        super().__init__(f"Template {self.template_name} is not configured for this Sms client.")

__init__(template_name)

Initialize the error with the missing template name.

Parameters:

Name	Type	Description	Default
template_name	str	The template identifier not configured for this client.	required

Source code in src/notificator/infra/sms_clients/exceptions.py

def __init__(self, template_name: str) -> None:
    """Initialize the error with the missing template name.

    Args:
        template_name: The template identifier not configured for this client.
    """
    self.template_name = template_name
    super().__init__(f"Template {self.template_name} is not configured for this Sms client.")

TemplateVersionNotAvailableError

Bases: SmsNotificationError

Raised when a template version is not available.

Source code in src/notificator/infra/sms_clients/exceptions.py

class TemplateVersionNotAvailableError(SmsNotificationError):
    """Raised when a template version is not available."""

    def __init__(self, template_name: str, version: str) -> None:
        """Initialize the error with the missing template version.

        Args:
            template_name: Template identifier requested by the caller.
            version: Template version that could not be resolved.
        """
        self.template_name = template_name
        self.version = version
        super().__init__(
            f"Version {self.version} is not available for template {self.template_name}."
        )

__init__(template_name, version)

Initialize the error with the missing template version.

Parameters:

Name	Type	Description	Default
template_name	str	Template identifier requested by the caller.	required
version	str	Template version that could not be resolved.	required

Source code in src/notificator/infra/sms_clients/exceptions.py

def __init__(self, template_name: str, version: str) -> None:
    """Initialize the error with the missing template version.

    Args:
        template_name: Template identifier requested by the caller.
        version: Template version that could not be resolved.
    """
    self.template_name = template_name
    self.version = version
    super().__init__(
        f"Version {self.version} is not available for template {self.template_name}."
    )

TwilioMissingSenderIdError

Bases: SmsNotificationError

Raised when a twilio client is missing a sender identity.

Source code in src/notificator/infra/sms_clients/exceptions.py

class TwilioMissingSenderIdError(SmsNotificationError):
    """Raised when a twilio client is missing a sender identity."""

    def __init__(self) -> None:
        """Initialize the error for a missing sender identity."""
        super().__init__(
            "Twilio client is missing a sender identity. Either provide a valid `sender_phone_number` "
            "or a `messaging_service_sid`."
        )

__init__()

Initialize the error for a missing sender identity.

Source code in src/notificator/infra/sms_clients/exceptions.py

def __init__(self) -> None:
    """Initialize the error for a missing sender identity."""
    super().__init__(
        "Twilio client is missing a sender identity. Either provide a valid `sender_phone_number` "
        "or a `messaging_service_sid`."
    )