Python aiosmtplib 异步发送 Email 介绍

在 上张贴 由 cx_sun发表回复

aiosmtplib 是一个专为 Python asyncio 设计的异步 SMTP 客户端库，用于高效发送电子邮件。它基于标准 SMTP 协议，支持 TLS/SSL 加密、STARTTLS 升级、多部分消息（如纯文本 + HTML），并无缝集成异步事件循环。相比同步库 smtplib，aiosmtplib 避免阻塞 I/O，适合高并发场景，如 Web 服务或批量邮件任务。

关键特性

  • 异步操作：所有连接、认证和发送均使用 await 关键字，支持 asyncio.gather() 并行处理多个客户端实例（SMTP 协议本身是顺序的，不宜在单一连接上并发发送）。
  • 消息格式：兼容 email.message.EmailMessage 和 MIME 子类，支持附件、HTML 等。
  • 加密支持：自动或手动处理 STARTTLS 和 SSL/TLS。
  • 安装：pip install aiosmtplib（若需代理，额外 pip install python-socks）。
  • 局限：官方不直接支持代理，但可通过自定义 socket（如 SOCKS 代理）扩展。

典型流程：

  1. 创建 SMTP 实例（可选预设主机/端口）。
  2. await connect() 建立连接（或用上下文管理器 async with SMTP(...) as client: 自动管理）。
  3. 构建 EmailMessage 对象。
  4. await send_message(message) 发送。
  5. await quit() 关闭（上下文管理器自动处理）。

基本用法示例（无代理）

以下是简单异步发送的代码示例，使用 Gmail SMTP（需启用“应用专用密码”）：

import asyncio
from email.message import EmailMessage
import aiosmtplib

async def send_email():
    message = EmailMessage()
    message["From"] = "[email protected]"
    message["To"] = "[email protected]"
    message["Subject"] = "异步测试邮件"
    message.set_content("这是使用 aiosmtplib 发送的纯文本邮件！")

    # 创建客户端
    smtp_client = aiosmtplib.SMTP(
        hostname="smtp.gmail.com",
        port=587,
        start_tls=True,  # 启用 STARTTLS
        username="[email protected]",
        password="your_app_password"
    )

    async with smtp_client:  # 自动连接和关闭
        await smtp_client.send_message(message)
        print("邮件发送成功！")

asyncio.run(send_email())
  • 多部分消息（纯文本 + HTML）：
  from email.mime.multipart import MIMEMultipart
  from email.mime.text import MIMEText

  msg = MIMEMultipart("alternative")
  msg["From"] = "[email protected]"
  msg["To"] = "[email protected]"
  msg["Subject"] = "HTML 测试"
  msg.attach(MIMEText("纯文本版本", "plain"))
  msg.attach(MIMEText("<h1>HTML 版本</h1>", "html"))
  await smtp_client.send_message(msg)

代理（Proxy）使用

aiosmtplib 官方不内置代理支持，但可以通过 python_socks 库创建自定义 socket（SOCKS4/5 代理），然后传递给 SMTP(sock=sock)。这允许邮件流量通过代理隧道发送，常用于绕过网络限制或增强隐私。

  • 依赖：pip install python-socks（异步 SOCKS 支持）。
  • 步骤：
  1. 配置代理（主机、端口、类型、凭证）。
  2. 使用 Proxy 创建异步 socket：await proxy.connect(dest_host, dest_port)。
  3. 初始化 SMTP 时传入 sock=sock，并指定 start_tls=True 等。
  4. 其余发送逻辑不变。

完整示例（含 SOCKS 代理，支持环境变量配置）

基于提供的附件文件（email_sender2.py），以下是封装类实现的示例。它使用 dataclasses 管理配置、dotenv 加载环境变量，并可选启用代理。假设 .env 文件：

[email protected]
EMAIL_PASSWORD=your_app_password
EMAIL_PROXY_HOST=proxy.example.com
EMAIL_PROXY_PORT=1080
EMAIL_PROXY_USER=proxy_user  # 可选
EMAIL_PROXY_PASS=proxy_pass  # 可选
import asyncio
import logging
import os
from dataclasses import dataclass
from email.message import EmailMessage
from typing import Optional

import aiosmtplib
from dotenv import load_dotenv
from python_socks import ProxyType
from python_socks.async_.asyncio import Proxy

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class EmailConfig:
    """邮件配置。"""
    sender_email: str
    sender_password: str
    smtp_host: str = "smtp.gmail.com"
    smtp_port: int = 587

    @classmethod
    def from_env(cls) -> "EmailConfig":
        load_dotenv()
        sender_email = os.getenv("EMAIL_SENDER")
        sender_password = os.getenv("EMAIL_PASSWORD")
        if not sender_email or not sender_password:
            raise ValueError("缺少 EMAIL_SENDER 或 EMAIL_PASSWORD")
        return cls(
            sender_email=sender_email,
            sender_password=sender_password,
            smtp_host=os.getenv("EMAIL_SMTP_SERVER", "smtp.gmail.com"),
            smtp_port=int(os.getenv("EMAIL_SMTP_PORT", "587")),
        )

@dataclass
class ProxyConfig:
    """代理配置（SOCKS5 默认）。"""
    host: str
    port: int
    proxy_type: ProxyType = ProxyType.SOCKS5
    username: Optional[str] = None
    password: Optional[str] = None

    @classmethod
    def from_env(cls) -> Optional["ProxyConfig"]:
        host = os.getenv("EMAIL_PROXY_HOST")
        port_str = os.getenv("EMAIL_PROXY_PORT")
        if not host or not port_str:
            return None
        return cls(
            host=host,
            port=int(port_str),
            username=os.getenv("EMAIL_PROXY_USER"),
            password=os.getenv("EMAIL_PROXY_PASS"),
        )

class EmailSender:
    """异步邮件发送器，支持 SOCKS 代理。"""

    def __init__(self, email_config: EmailConfig, proxy_config: Optional[ProxyConfig] = None):
        self.email_config = email_config
        self.proxy_config = proxy_config

    async def _create_smtp_connection(self) -> aiosmtplib.SMTP:
        """创建 SMTP 连接（代理可选）。"""
        if self.proxy_config:
            logger.info(f"通过代理 {self.proxy_config.host}:{self.proxy_config.port} 连接 SMTP")
            proxy = Proxy(
                proxy_type=self.proxy_config.proxy_type,
                host=self.proxy_config.host,
                port=self.proxy_config.port,
                username=self.proxy_config.username,
                password=self.proxy_config.password,
            )
            sock = await proxy.connect(
                dest_host=self.email_config.smtp_host,
                dest_port=self.email_config.smtp_port,
            )
            return aiosmtplib.SMTP(
                start_tls=True,
                username=self.email_config.sender_email,
                password=self.email_config.sender_password,
                sock=sock,  # 关键：传入自定义 socket
            )
        else:
            logger.info("直接连接 SMTP（无代理）")
            return aiosmtplib.SMTP(
                hostname=self.email_config.smtp_host,
                port=self.email_config.smtp_port,
                start_tls=True,
                username=self.email_config.sender_email,
                password=self.email_config.sender_password,
            )

    async def send_email(self, recipient: str, subject: str, body: str) -> None:
        """发送邮件。"""
        message = EmailMessage()
        message["From"] = self.email_config.sender_email
        message["To"] = recipient
        message["Subject"] = subject
        message.set_content(body)

        smtp_client = await self._create_smtp_connection()
        try:
            await smtp_client.connect()
            await smtp_client.send_message(message)
            logger.info(f"邮件发送成功至 {recipient}")
        except Exception as e:
            logger.error(f"发送失败: {e}")
            raise
        finally:
            await smtp_client.quit()

# 使用示例
async def main():
    email_config = EmailConfig.from_env()
    proxy_config = ProxyConfig.from_env()  # 如果 .env 无代理配置，则为 None

    sender = EmailSender(email_config, proxy_config)
    await sender.send_email(
        recipient="[email protected]",
        subject="代理测试邮件",
        body="这是通过 SOCKS 代理异步发送的邮件！"
    )
    print("完成！")

if __name__ == "__main__":
    asyncio.run(main())

运行与注意

  • 执行 python script.py，日志会显示是否使用代理（e.g., “通过代理 proxy.example.com:1080 连接 SMTP”）。
  • 错误处理：捕获 SMTPException 或通用 Exception，检查网络/凭证。
  • 性能：批量发送时，创建多个 EmailSender 实例并用 asyncio.gather() 并行。
  • 扩展：添加 CC/BCC 用 message["Cc"] = ", ".join(cc_list)；附件用 message.add_attachment()。

此实现直接参考附件文件的核心逻辑，便于生产使用。若需更多功能（如重试机制），可进一步扩展。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注