Amazon SNS — boto3 in Python¶

import boto3

sns = boto3.client("sns", region_name="us-east-1")

std = sns.create_topic(
    Name="order-events",
    Tags=[{"Key": "env", "Value": "test"}],
    Attributes={"DisplayName": "Orders"},
)["TopicArn"]

fifo = sns.create_topic(
    Name="order-events.fifo",                                         # suffix is mandatory for FIFO
    Attributes={"FifoTopic": "true", "ContentBasedDeduplication": "true"},
)["TopicArn"]

print(std)     # arn:aws:sns:us-east-1:123456789012:order-events
print(fifo)    # arn:aws:sns:us-east-1:123456789012:order-events.fifo
print(sns.create_topic(Name="order-events")["TopicArn"] == std)       # True: create_topic is idempotent
print(sns.get_topic_attributes(TopicArn=fifo)["Attributes"]["FifoTopic"])   # true
print(sns.list_tags_for_resource(ResourceArn=std)["Tags"])            # [{'Key': 'env', 'Value': 'test'}]

import json

sqs = boto3.client("sqs", region_name="us-east-1")
url = sqs.create_queue(QueueName="billing")["QueueUrl"]
queue_arn = sqs.get_queue_attributes(QueueUrl=url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"]
# (queue policy allowing the topic: see 01 Core Concepts or the minimal example in the overview)

subscription_arn = sns.subscribe(
    TopicArn=std,
    Protocol="sqs",
    Endpoint=queue_arn,
    ReturnSubscriptionArn=True,                                       # real ARN even while pending
    Attributes={
        "RawMessageDelivery": "true",
        "FilterPolicy": json.dumps({"event_type": ["created"]}),      # must be a JSON *string*
        "FilterPolicyScope": "MessageAttributes",                     # or "MessageBody"
    },
)["SubscriptionArn"]

sns.set_subscription_attributes(                                      # change one attribute later
    SubscriptionArn=subscription_arn,
    AttributeName="FilterPolicy",
    AttributeValue=json.dumps({"event_type": ["created", "paid"]}),
)
print(sns.get_subscription_attributes(SubscriptionArn=subscription_arn)["Attributes"]["FilterPolicy"])
# {"event_type": ["created", "paid"]}

for page in sns.get_paginator("list_subscriptions_by_topic").paginate(TopicArn=std):
    for sub in page["Subscriptions"]:
        print(sub["Protocol"], sub["Endpoint"])
# sqs arn:aws:sqs:us-east-1:123456789012:billing

topics = [t["TopicArn"] for page in sns.get_paginator("list_topics").paginate() for t in page["Topics"]]

import json

response = sns.publish(
    TopicArn=std,
    Message=json.dumps({"order_id": "o1"}),               # the payload; SNS does not interpret it
    Subject="order created",                              # optional, < 100 chars; absent from raw deliveries
    MessageAttributes={
        "event_type":   {"DataType": "String",       "StringValue": "created"},
        "amount_cents": {"DataType": "Number",       "StringValue": "1250"},
        "channels":     {"DataType": "String.Array", "StringValue": json.dumps(["web", "mobile"])},
    },
)
print(sorted(k for k in response if k != "ResponseMetadata"))     # ['MessageId']

response = sns.publish_batch(
    TopicArn=std,
    PublishBatchRequestEntries=[
        {"Id": "a", "Message": "1", "MessageAttributes": {"event_type": {"DataType": "String", "StringValue": "created"}}},
        {"Id": "b", "Message": "2"},
    ],
)
print([x["Id"] for x in response["Successful"]], response["Failed"])      # ['a', 'b'] []

# app/publisher.py: chunks of 10, raises when anything failed (itertools.batched needs Python 3.12+)
from itertools import batched

for chunk in batched(events, 10):
    entries = [{"Id": str(i), **build_publish_kwargs(e)} for i, e in enumerate(chunk)]
    response = sns.publish_batch(TopicArn=topic_arn, PublishBatchRequestEntries=entries)
    if response["Failed"]:
        raise PublishError(response["Failed"])

r = sns.publish(
    TopicArn=fifo,
    Message="order paid",
    MessageGroupId="order-1",                  # ordering key: one group per order, customer, device...
    # MessageDeduplicationId="order-1-paid",   # required unless ContentBasedDeduplication is on (it is, above)
)

sns.publish(
    TopicArn=std,
    MessageStructure="json",
    Message=json.dumps({"default": "order o1 created", "sqs": json.dumps({"order_id": "o1"}), "email": "Your order o1 is confirmed"}),
)

sns.set_topic_attributes(TopicArn=std, AttributeName="DisplayName", AttributeValue="Orders")
sns.set_topic_attributes(TopicArn=std, AttributeName="KmsMasterKeyId", AttributeValue="alias/aws/sns")   # SSE
sns.set_topic_attributes(TopicArn=std, AttributeName="MaximumMessageSize", AttributeValue="1048576")    # up to 1 MiB, SQS/Lambda/Firehose only

sns.add_permission(                                                   # adds a statement to the topic policy
    TopicArn=std, Label="partner-publish", AWSAccountId=["111122223333"], ActionName=["Publish"],
)

from botocore.exceptions import ClientError


def publish_or_none(sns, topic_arn: str, **kwargs) -> str | None:
    """None when the topic is gone; everything else is raised."""
    try:
        return sns.publish(TopicArn=topic_arn, **kwargs)["MessageId"]
    except sns.exceptions.NotFoundException:
        return None
    except ClientError as exc:
        if exc.response["Error"]["Code"] in {"Throttled", "InternalError"}:
            raise RuntimeError("SNS is busy; retry later") from exc
        raise

import random
import time

from botocore.exceptions import ClientError

RETRYABLE = {"Throttled", "InternalError", "KMSThrottling"}


def call_with_retry(fn, *, attempts: int = 5, base: float = 0.2, sleep=time.sleep):
    for attempt in range(1, attempts + 1):
        try:
            return fn()
        except ClientError as exc:
            if exc.response["Error"]["Code"] not in RETRYABLE or attempt == attempts:
                raise
            sleep(random.uniform(0, base * 2 ** attempt))          # full jitter


call_with_retry(lambda: sns.publish(TopicArn=std, Message="m"))