如何在AWS SNS主题中向特定订阅者发送邮件（Django API场景）

1. 初始化 AWS SNS 客户端 #

先在Django项目中配置boto3，建议通过环境变量或IAM角色（部署在AWS服务上时）管理凭证，避免硬编码：

import boto3
import json
from django.conf import settings

class SNSMailHandler:
    def __init__(self):
        self.sns_client = boto3.client(
            'sns',
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name=settings.AWS_SNS_REGION
        )
        self.target_topic_arn = settings.AWS_SNS_TOPIC_ARN  # 你的SNS主题ARN

2. 获取主题下所有有效邮箱订阅者 #

SNS的list_subscriptions_by_topic接口返回分页结果，需循环获取全部订阅者，并筛选出已确认的邮箱订阅：

def fetch_all_email_subscribers(self):
    subscribers = []
    next_token = None
    while True:
        request_params = {'TopicArn': self.target_topic_arn}
        if next_token:
            request_params['NextToken'] = next_token
        
        response = self.sns_client.list_subscriptions_by_topic(**request_params)
        # 过滤出已确认的邮箱订阅者
        valid_email_subs = [
            sub for sub in response['Subscriptions']
            if sub['Protocol'] == 'email' and sub['SubscriptionArn'] != 'PendingConfirmation'
        ]
        subscribers.extend(valid_email_subs)
        
        next_token = response.get('NextToken')
        if not next_token:
            break
    return subscribers

3. 匹配邮箱或创建新订阅 #

检查用户邮箱是否已订阅，未订阅则创建新的邮箱订阅：

def get_or_create_subscription(self, user_email):
    existing_subs = self.fetch_all_email_subscribers()
    # 匹配目标邮箱
    matched_sub = next(
        (sub for sub in existing_subs if sub['Endpoint'] == user_email),
        None
    )
    if matched_sub:
        return matched_sub['SubscriptionArn']
    
    # 创建新订阅
    create_response = self.sns_client.subscribe(
        TopicArn=self.target_topic_arn,
        Protocol='email',
        Endpoint=user_email
    )
    # 注意：新邮箱订阅默认处于待确认状态，用户点击邮件中的链接后才会生效
    return create_response['SubscriptionArn']

4. 调用 publish 发送定向邮件 #

通过订阅者ARN定向发送邮件，支持纯文本与HTML格式：

def send_user_email(self, user_email, subject, content):
    subscription_arn = self.get_or_create_subscription(user_email)
    
    # 构造多格式消息结构，SNS会根据协议自动适配
    message_content = {
        'default': content,
        'email': f"Subject: {subject}\n\n{content}",
        'email-json': json.dumps({"subject": subject, "content": content})
    }
    
    try:
        publish_response = self.sns_client.publish(
            TargetArn=subscription_arn,
            Message=json.dumps(message_content),
            MessageStructure='json'
        )
        return publish_response['MessageId']
    except self.sns_client.exceptions.InvalidParameterException:
        # 处理订阅未确认、无效ARN等问题
        raise ValueError("订阅未确认或ARN无效，请检查用户邮箱状态")
    except Exception as e:
        # 捕获其他SNS服务异常
        raise e

关键注意事项 #

• 凭证安全：生产环境优先使用IAM角色（如EC2实例角色、Lambda执行角色），避免在代码中硬编码AWS密钥。
• 订阅确认：邮箱订阅默认需要用户手动确认，强行跳过确认不符合AWS最佳实践，可能触发垃圾邮件检测机制。
• 分页处理：必须处理list_subscriptions_by_topic的分页逻辑，否则订阅者数量超过100时会丢失数据。
• 异常捕获：针对SNS的特定异常（如权限不足、参数错误）做针对性处理，提升接口健壮性。

内容的提问来源于stack exchange，提问作者Hetvee Shah