Skip to content

SNS Notifier

Sends notifications via Amazon Simple Notification Service (SNS).

SNS topic notification delivery.

Provides notification delivery via Amazon Simple Notification Service (SNS).

SNSNotifier dataclass 

SNSNotifier()

Bases: Notifier[SNSTopicTarget]

Notifier implementation for publishing to Amazon SNS topics.

Publishes notification content to SNS topics for fanout to subscribers (email, SMS, HTTP endpoints, etc.).

Example 
notifier = SNSNotifier()
result = notifier.notify(
    content=NotificationContent(subject="Alert", message="Critical event"),
    target=SNSTopicTarget(topic="arn:aws:sns:us-east-1:123456789012:my-topic")
)

notify 

notify(content, target)

Publish a notification to an SNS topic.

Parameters:

Name Type Description Default
content NotificationContent

The notification content including subject and message.

 required
target SNSTopicTarget

The SNS topic target containing the topic ARN.

 required

Returns:

Type Description
NotifierResult

Result indicating success or failure with response details.
Source code in src/aibs_informatics_aws_lambda/handlers/notifications/notifiers/sns.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def notify(self, content: NotificationContent, target: SNSTopicTarget) -> NotifierResult:
    """Publish a notification to an SNS topic.

    Args:
        content (NotificationContent): The notification content including subject and message.
        target (SNSTopicTarget): The SNS topic target containing the topic ARN.

    Returns:
        Result indicating success or failure with response details.
    """
    try:
        response = publish_to_topic(
            message=content.message,
            subject=content.subject,
            topic_arn=target.topic,
        )
        return NotifierResult(
            response=json.dumps(response),
            success=(200 <= response["ResponseMetadata"]["HTTPStatusCode"] < 300),
            target=target.to_dict(),
        )
    except Exception as e:
        return NotifierResult(
            target=target.to_dict(),
            success=False,
            response=str(e),
        )