# 方法1：通过环境变量（推荐）
export AWS_ACCESS_KEY_ID='AKIAXXXXXXXXXXXXXXXX'
export AWS_SECRET_ACCESS_KEY='wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY'

# 方法2：创建~/.aws/credentials文件
[default]
aws_access_key_id = AKIAXXXXXXXXXXXXXXXX
aws_secret_access_key = wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY

import boto3

def send_text_email():
    client = boto3.client('ses', region_name='us-east-1')
    
    response = client.send_email(
        Source='noreply@example.com',
        Destination={
            'ToAddresses': ['recipient@gmail.com']
        },
        Message={
            'Subject': {
                'Data': '测试邮件主题'
            },
            'Body': {
                'Text': {
                    'Data': '这是一封测试邮件的正文内容。'
                }
            }
        }
    )
    print(f"邮件发送成功！MessageID: {response['MessageId']}")

send_text_email()

def send_html_email():
    client = boto3.client('ses', region_name='us-east-1')
    
    html_content = """
    <html>
    <head></head>
    <body>
        <h1>欢迎使用我们的服务</h1>
        <p>点击<a href="https://example.com" rel="external nofollow" >这里</a>激活账户</p>
    </body>
    </html>
    """
    
    response = client.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': ['recipient@gmail.com']},
        Message={
            'Subject': {'Data': 'HTML格式测试邮件'},
            'Body': {
                'Html': {'Data': html_content}
            }
        }
    )
    print(f"邮件发送成功！MessageID: {response['MessageId']}")

现代邮件客户端会自动选择支持的格式显示。建议始终提供文本版本作为备选：

def send_multipart_email():
    client = boto3.client('ses', region_name='us-east-1')
    
    response = client.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': ['recipient@gmail.com']},
        Message={
            'Subject': {'Data': '多部分格式测试邮件'},
            'Body': {
                'Text': {'Data': '如果看到这行文字，说明您的客户端不支持HTML格式'},
                'Html': {'Data': '<strong>HTML内容</strong>会自动显示'}
            }
        }
    )

当邮件内容需要动态生成时，模板引擎可大幅提升开发效率：

from jinja2 import Environment, FileSystemLoader

# 创建模板环境
env = Environment(loader=FileSystemLoader('templates'))

def send_templated_email(username):
    client = boto3.client('ses')
    
    # 加载并渲染模板
    template = env.get_template('welcome.html')
    html_body = template.render(username=username)
    
    response = client.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': ['recipient@gmail.com']},
        Message={
            'Subject': {'Data': f'欢迎，{username}！'},
            'Body': {'Html': {'Data': html_body}}
        }
    )

通过send_raw_email()方法可发送包含附件的复杂邮件：

import email
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication

def send_email_with_attachment():
    msg = MIMEMultipart()
    msg['Subject'] = '带附件的测试邮件'
    msg['From'] = 'noreply@example.com'
    msg['To'] = 'recipient@gmail.com'
    
    # 添加文本正文
    text_part = MIMEText('这是邮件正文内容')
    msg.attach(text_part)
    
    # 添加PDF附件
    with open('report.pdf', 'rb') as f:
        pdf_part = MIMEApplication(f.read(), _subtype='pdf')
        pdf_part.add_header('Content-Disposition', 'attachment', filename='report.pdf')
        msg.attach(pdf_part)
    
    # 发送原始邮件
    client = boto3.client('ses')
    response = client.send_raw_email(
        Source=msg['From'],
        Destinations=[msg['To']],
        RawMessage={'Data': msg.as_string()}
    )

当需要发送大量邮件时，可采用以下策略：

import concurrent.futures

def batch_send_emails(recipients):
    def send_single(recipient):
        client = boto3.client('ses')
        client.send_email(
            Source='noreply@example.com',
            Destination={'ToAddresses': [recipient]},
            Message={
                'Subject': {'Data': '批量发送测试'},
                'Body': {'Text': {'Data': f'尊敬的{recipient}，这是批量邮件测试'}}
            }
        )
    
    # 使用线程池并发发送
    with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(send_single, recipients)

错误现象：InvalidClientTokenId或Security token included in the request is invalid

解决方案：

检查环境变量是否正确设置
确认~/.aws/credentials文件权限为600
使用AWS CLI测试密钥有效性：

aws ses verify-email-identity --email-address noreply@example.com

错误现象：Email address is not verified或Daily quota exceeded

解决方案：

确保所有收件人邮箱已验证

监控发送配额：

def check_quota():
    client = boto3.client('ses')
    quota = client.get_send_quota()
    print(f"24小时内已发送: {quota['SentLast24Hours']}/{quota['Max24HourSend']}")

优化建议：

配置SPF和DKIM记录（在域名DNS设置中添加TXT记录）

设置退订链接（通过SES的`FeedbackForwardingEnabled`参数）

避免使用过多营销词汇和全大写字母

频繁创建SES客户端会消耗资源，建议使用单例模式：

from functools import lru_cache

@lru_cache(maxsize=1)
def get_ses_client():
    return boto3.client('ses', region_name='us-east-1')

# 使用方式
client = get_ses_client()

对于Web应用，可使用Celery等任务队列实现异步发送：

from celery import Celery

app = Celery('email_tasks', broker='redis://localhost:6379/0')

@app.task
def async_send_email(to, subject, body):
    client = boto3.client('ses')
    client.send_email(
        Source='noreply@example.com',
        Destination={'ToAddresses': [to]},
        Message={'Subject': {'Data': subject}, 'Body': {'Text': {'Data': body}}}
    )

通过CloudWatch监控SES指标，设置阈值告警：

def monitor_ses_metrics():
    cloudwatch = boto3.client('cloudwatch')
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/SES',
        MetricName='Send',
        Dimensions=[{'Name': 'Region', 'Value': 'us-east-1'}],
        Period=3600,
        Statistics=['Sum'],
        StartTime=datetime.utcnow() - timedelta(hours=1),
        EndTime=datetime.utcnow()
    )
    print(f"过去1小时发送量: {response['Datapoints'][0]['Sum'] if response['Datapoints'] else 0}")

以下是一个完整的邮件服务类实现，封装了常用功能：

import boto3
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from typing import Union, List, Optional

class EmailService:
    def __init__(self, region: str = 'us-east-1'):
        self.client = boto3.client('ses', region_name=region)
    
    def send_text(
        self,
        to: Union[str, List[str]],
        subject: str,
        body: str,
        from_addr: Optional[str] = None
    ) -> str:
        """发送纯文本邮件"""
        if isinstance(to, str):
            to = [to]
        
        response = self.client.send_email(
            Source=from_addr or 'noreply@example.com',
            Destination={'ToAddresses': to},
            Message={
                'Subject': {'Data': subject},
                'Body': {'Text': {'Data': body}}
            }
        )
        return response['MessageId']
    
    def send_html(
        self,
        to: Union[str, List[str]],
        subject: str,
        html: str,
        from_addr: Optional[str] = None
    ) -> str:
        """发送HTML邮件"""
        if isinstance(to, str):
            to = [to]
        
        response = self.client.send_email(
            Source=from_addr or 'noreply@example.com',
            Destination={'ToAddresses': to},
            Message={
                'Subject': {'Data': subject},
                'Body': {'Html': {'Data': html}}
            }
        )
        return response['MessageId']
    
    def send_raw(
        self,
        to: Union[str, List[str]],
        subject: str,
        text: Optional[str] = None,
        html: Optional[str] = None,
        attachments: Optional[List[dict]] = None,
        from_addr: Optional[str] = None
    ) -> str:
        """发送包含附件的复杂邮件"""
        msg = MIMEMultipart()
        msg['Subject'] = subject
        msg['From'] = from_addr or 'noreply@example.com'
        msg['To'] = to[0] if isinstance(to, list) else to
        
        # 添加正文
        if text:
            msg.attach(MIMEText(text, 'plain'))
        if html:
            msg.attach(MIMEText(html, 'html'))
        
        # 添加附件
        if attachments:
            for att in attachments:
                with open(att['path'], 'rb') as f:
                    part = MIMEApplication(f.read(), _subtype=att['type'])
                    part.add_header('Content-Disposition', 'attachment', filename=att['name'])
                    msg.attach(part)
        
        # 发送邮件
        destinations = [to] if isinstance(to, str) else to
        response = self.client.send_raw_email(
            Source=msg['From'],
            Destinations=destinations,
            RawMessage={'Data': msg.as_string()}
        )
        return response['MessageId']

# 使用示例
if __name__ == '__main__':
    service = EmailService()
    
    # 发送纯文本邮件
    service.send_text(
        to='recipient@gmail.com',
        subject='测试文本邮件',
        body='这是一封测试邮件的正文内容。'
    )
    
    # 发送HTML邮件
    service.send_html(
        to=['user1@example.com', 'user2@example.com'],
        subject='测试HTML邮件',
        html='<h1>欢迎使用</h1><p>点击<a href="#" rel="external nofollow" >这里</a>激活账户</p>'
    )
    
    # 发送带附件的邮件
    service.send_raw(
        to='recipient@gmail.com',
        subject='测试附件邮件',
        text='这是邮件正文',
        attachments=[
            {'path': 'report.pdf', 'name': '年度报告.pdf', 'type': 'pdf'}
        ]
    )

通过本文的实践指南，开发者可以快速掌握使用Python和AWS SES发送邮件的核心技能。从基础功能到高级特性，从错误处理到性能优化，覆盖了实际开发中的常见场景。

未来邮件服务的发展趋势包括：

AI驱动的个性化内容：基于用户行为动态生成邮件内容

增强的安全性：更严格的反垃圾邮件机制和隐私保护

无服务器架构：结合Lambda实现完全自动化的邮件流水线

建议开发者持续关注AWS官方文档，及时了解SES的新功能更新，特别是关于发送配额提升和反垃圾邮件政策的调整。通过合理利用云服务，可以构建出既高效又可靠的邮件发送系统，为业务发展提供有力支持。

以上就是Python调用Boto库实现邮件自动化发送的完整指南的详细内容，更多关于Python自动发送邮件的资料请关注风君子博客其它相关文章！

您可能感兴趣的文章:

使用Python实现自动发送邮件的实战指南(以QQ/163为例)

使用Python自动发送邮件的实现步骤

从入门到精通详解Python自动化邮件发送的完全指南

Python实现批量发送邮件脚本

Python3实现SMTP发送邮件的实战指南

基于Python+smtplib实现邮件自动发送功能

华夏福临门盛世加强版怎么样？产品责任是重中之重

国寿福庆典版值得买吗？附投保建议

理财中低风险是几级？中低风险理财会亏吗

瑞泰泰安心防癌险优缺点是什么？这份分析不容错过

2020最新自助免费申请Office365教育版，免费5TOneDrive云盘详细图文教程

MacBook怎么绕过BootCamp安装Win10双系统？

国内知名的前端博客

域名为什么会被墙？如何检测域名是否被墙？域名被墙如何处理？

最新的windows xp sp3产品密匙 xp sp3序列号 xp sp3激活码

Python调用Boto库实现邮件自动化发送的完整指南

目录

访问AWS官网注册账号，选择免费套餐（Free Tier）即可开始使用SES。

AWS官方推荐使用boto3（新一代SDK），但部分旧项目仍在使用boto（v2版本）。本文以boto3为例演示： pip install boto3

错误现象：InvalidClientTokenId或Security token included in the request is invalid 解决方案： 检查环境变量是否正确设置 确认~/.aws/credentials文件权限为600 使用AWS CLI测试密钥有效性： aws ses verify-email-identity --email-address noreply@example.com

优化建议： 配置SPF和DKIM记录（在域名DNS设置中添加TXT记录） 设置退订链接（通过SES的FeedbackForwardingEnabled参数） 避免使用过多营销词汇和全大写字母

频繁创建SES客户端会消耗资源，建议使用单例模式： from functools import lru_cache @lru_cache(maxsize=1) def get_ses_client(): return boto3.client('ses', region_name='us-east-1') # 使用方式 client = get_ses_client()

More posts

站内搜索

标签云

热门文章

友情链接

AWS官方推荐使用`boto3`（新一代SDK），但部分旧项目仍在使用`boto`（v2版本）。本文以`boto3`为例演示：

pip install boto3

错误现象：`InvalidClientTokenId`或`Security token included in the request is invalid`

解决方案：

检查环境变量是否正确设置

确认`~/.aws/credentials`文件权限为600

使用AWS CLI测试密钥有效性：

aws ses verify-email-identity --email-address noreply@example.com

优化建议：

配置SPF和DKIM记录（在域名DNS设置中添加TXT记录）

设置退订链接（通过SES的`FeedbackForwardingEnabled`参数）

避免使用过多营销词汇和全大写字母

频繁创建SES客户端会消耗资源，建议使用单例模式：

from functools import lru_cache @lru_cache(maxsize=1) def get_ses_client(): return boto3.client('ses', region_name='us-east-1') # 使用方式 client = get_ses_client()