API Reference

Base Notifier

This module contains the base notifier class which is the boilerplate for all notifiers

class notihub.base_notifier.BaseNotifier[source]

Bases: ABC

Base class for all notifiers used to register them

__init__()
_abc_impl = <_abc._abc_data object>
abstract send_email_notification(*, subject, email_data, recipients, sender, template, cc_emails=None, bcc_emails=None, **kwargs)[source]

Sends an email notification to the given email

Return type:

str

abstract send_push_notification(device, message, **kwargs)[source]

Sends a push notification to the given message

Return type:

str

abstract send_sms_notification(phone_number, message, **kwargs)[source]

Sends a SMS notification to the given phone number

Return type:

str

AWS Notifier

This module contains the AWS notifier class which is used to send notifications via AWS SNS, SES or Pinpoint

class notihub.notifiers.aws.notifier.AWSNotifier(aws_access_key_id=None, aws_secret_access_key=None, region_name=None)[source]

Bases: SNSClient, SESClient, PinpointClient, BaseNotifier

Centralized class to send notifications via AWS SNS, SES or Pinpoint using class inheritance to initialize the clients.

__init__(aws_access_key_id=None, aws_secret_access_key=None, region_name=None)
__post_init__()[source]

Initializes the parent client classes after AWSNotifier’s own fields are set. This ensures that self.sns_client, self.ses_client, and self.pinpoint_client are created by their respective parent classes.

_abc_impl = <_abc._abc_data object>
_build_apns_message(*, title, body, action, deep_link_url, image_url, silent_push, custom_data)

Build the APNS message.

Return type:

Dict[str, Any]

_build_gcm_message(*, title, body, action, deep_link_url, image_url, silent_push, custom_data, time_to_live, priority)
Return type:

Dict[str, Any]

_prepare_custom_data(custom_data, *, deep_link_url, image_url)

Return a copy of custom_data enriched with deeplink / image_url.

Return type:

Dict[str, Any]

aws_access_key_id: str = None
aws_secret_access_key: str = None
create_device_endpoint(platform_application_arn, device_token, custom_user_data='', **kwargs)

Creates a platform endpoint for the given device token.

Parameters:

  • platform_application_arn (str) – The ARN of the platform application

  • (e.g.

  • APNS).

  • device_token (str) – The token associated with the device to register.

  • custom_user_data (str, optional) – The custom user data to associate with the

  • string (device endpoint. This should be a JSON-formatted)

  • data (representing user-specific)

  • ID (such as user)

  • type (subscription)

  • etc.

  • provided (If not)

  • endpoint. (no custom user data is associated with the)

  • "". (Defaults to)

Returns:

Response from the SNS client operation,

which includes the platform endpoint details or an error message if the operation fails.

Return type:

dict

create_email_template(template_name, subject, text_body, html_body)

Creates an email template with the given name

Parameters:

  • template_name (str) – The name of the template

  • subject (str) – The subject of the template

  • text_body (str) – The text body of the template

  • html_body (str) – The HTML body of the template

Returns:

Response of the client operation

Return type:

dict

create_pinpoint_endpoint(*, application_id, device_token, channel_type, user_id=None)

Creates a new Pinpoint endpoint with a generated unique ID. :type application_id: str :param application_id: The Pinpoint Application ID. :type device_token: str :param device_token: The push notification token from APNS/FCM. :type channel_type: str :param channel_type: Channel type (e.g., “GCM”, “APNS”, “APNS_SANDBOX”). :type user_id: Optional[str] :param user_id: The user ID of the application.

Return type:

Dict[str, Any]

Returns:

Dict containing API response and the generated ‘EndpointId’. Store the ‘EndpointId’ for future updates/deletions.

Raises:

ClientError – If the Pinpoint API call fails.

create_topic(topic_name)

Creates a topic with the given name

Parameters:

topic_name (str) – The name of the topic

Returns:

response of the client operation with the ARN of the topic

Return type:

dict

delete_device_endpoint(endpoint_arn, **kwargs)

Deletes the platform endpoint for the given endpoint ARN. :type endpoint_arn: str :param endpoint_arn: The ARN of the platform endpoint to delete. :type endpoint_arn: str

Returns:

Response from the SNS client operation, which includes the result of the delete operation or an error message if the operation fails.

Return type:

dict

delete_email_template(template_name)

Deletes an email template with the given name

Parameters:

template_name (str) – The name of the template

Returns:

Response of the client operation

Return type:

dict

delete_pinpoint_endpoint(*, application_id, endpoint_id)

Deletes a specific Pinpoint endpoint. :type application_id: str :param application_id: The Pinpoint Application ID. :type endpoint_id: str :param endpoint_id: The unique identifier of the endpoint to delete.

Return type:

Dict[str, Any]

Returns:

The response from the Pinpoint delete_endpoint operation.

Raises:

ClientError – If the Pinpoint API call fails.

delete_topic(topic_arn)

Deletes a topic with the given ARN

Parameters:

topic_arn (str) – The ARN of the topic

Returns:

Response of the client operation

Return type:

dict

get_email_template(template_name)

Gets an email template with the given name

Parameters:

template_name (str) – The name of the template

Returns:

Response with the template data

Return type:

dict

get_topic(topic_arn)

Gets a topic with the given ARN

Parameters:

topic_arn (str) – The ARN of the topic

Returns:

Response of the client operation

Return type:

dict

initialize_client(service_name=None)

Initialize the AWS client

Parameters:

service_name (str) – The name of the service

list_email_templates()

Lists all email templates

Parameters:

template_name (str) – The name of the template

Returns:

List of email templates

Return type:

list

region_name: str = None
send_email_notification(*, email_data, recipients, sender, template, cc_emails=None, bcc_emails=None, subject=None, **kwargs)

Sends an email notification to the given emails with a template

Parameters:

  • email_data (dict) – The data to be used in the email template

  • recipients (List[str]) – The recipients of the email

  • sender (str) – The sender of the email

  • template (str) – The name of the email template

Additional arguments:

subject (str): The subject of the email (not required if template is provided) cc_emails (List[str]): The CC emails of the email bcc_emails (List[str]): The BCC emails of the email *args: Additional arguments **kwargs: Additional keyword arguments

Returns:

Response of the client operation

Return type:

dict

send_pinpoint_push_notification(*, application_id, addresses, title, body, deep_link_url=None, image_url=None, custom_data=None, silent_push=False, time_to_live=None, priority=None, **kwargs)

Sends a push notification via Pinpoint to specified addresses. It groups addresses by service type (GCM, APNS) and sends a batch for each.

Return type:

List[Dict[str, Any]]

send_push_notification(device, message, title, payload=None, **kwargs)

Sends a push notification with a title to the given message

Parameters:

  • device (str) – The device to send the message to

  • title (str) – The title of the push notification

  • message (str) – The message to send

  • payload (dict, optional) – Custom payload to send. If not provided,

  • used. (a default will be)

Returns:

Response of the client operation

Return type:

dict

send_sms_notification(phone_number, message, **kwargs)

Sends a SMS notification to the given phone number

Parameters:

  • phone_number (str) – The phone number to send the message to

  • message (str) – The message to send

Additional arguments:

**kwargs: Additional keyword arguments

Returns:

Response of the client operation

Return type:

dict

send_topic_notification(*, topic_arn, message, subject, message_structure=None, **kwargs)

Sends a notification to the given topic

Parameters:

  • topic_arn (str) – The topic ARN

  • message (str) – The message to send

Additional arguments:

subject (str): The subject of the message target_arn (str): The target ARN message_structure (str): The message structure *args: Additional arguments **kwargs: Additional keyword arguments

Returns:

Response of the client operation

Return type:

dict

subscribe_to_topic(topic_arn, protocol, endpoint)

Subscribes the given endpoint to the given topic

Parameters:

  • topic_arn (str) – The topic ARN

  • protocol (str) – The protocol to use

  • endpoint (str) – The endpoint to subscribe to

Returns:

response of the client operation with the ARN of the subscription

Return type:

dict

update_device_endpoint(endpoint_arn, custom_user_data='', **kwargs)

Updates the CustomUserData for the given platform endpoint. :type endpoint_arn: str :param endpoint_arn: The ARN of the platform endpoint to update. :type endpoint_arn: str :type custom_user_data: str :param custom_user_data: The new custom user data to :type custom_user_data: str :param associate with the endpoint.: This should be a JSON-formatted string representing user-specific data.

Returns:

Response from the SNS client operation, which includes the updated platform endpoint details or an error message if the operation fails.

Return type:

dict

update_email_template(template_name, subject, text_body, html_body)

Updates an email template with the given name

Parameters:

  • template_name (str) – The name of the template

  • subject (str) – The subject of the template

  • text_body (str) – The text body of the template

  • html_body (str) – The HTML body of the template

Returns:

Response of the client operation

Return type:

dict

update_pinpoint_endpoint(*, application_id, endpoint_id, device_token=None, channel_type=None, user_id=None)

Updates specific attributes of an existing Pinpoint endpoint. :type application_id: str :param application_id: The Pinpoint Application ID. :type endpoint_id: str :param endpoint_id: The unique ID of the endpoint to update. :type device_token: Optional[str] :param device_token: Optional new push notification token. :type channel_type: Optional[str] :param channel_type: Optional channel type. :type user_id: Optional[str] :param user_id: Optional application’s user ID.

Return type:

Dict[str, Any]

Returns:

The response from the Pinpoint update_endpoint operation.

Raises:

ClientError – If the Pinpoint API call fails.

validate_credentials()

Validate AWS credentials are provided