基于Python和SQLAlchemy的高效邮件发送系统设计与实现

3. 邮件发送核心实现

3.1 基础邮件发送（SMTP）

我们使用 Python 的 smtplib 和 email 库实现邮件发送。这里要注意，生产环境建议使用环境变量管理敏感配置，而不是硬编码在代码里。

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

def send_email(to_email, subject, body, attachment_path=None):
    """发送邮件（支持附件）"""
    # 邮件服务器配置
    smtp_server = "smtp.qq.com"
    smtp_port = 465
    sender_email = "[email protected]"
    password = "your_smtp_password"

    # 创建邮件对象
    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['To'] = to_email
    msg['Subject'] = subject

    # 添加正文
    msg.attach(MIMEText(body, 'plain'))

    # 添加附件（如果有）
    if attachment_path:
        with open(attachment_path, "rb") as file:
            part = MIMEApplication(file.read(), Name=os.path.basename(attachment_path))
            part['Content-Disposition'] = f'attachment; filename="{os.path.basename(attachment_path)}"'
            msg.attach(part)

    # 发送邮件
    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(sender_email, password)
            server.sendmail(sender_email, to_email, msg.as_string())
        return True
    except Exception as e:
        print(f"邮件发送失败：{e}")
        return False

3.2 多收件人邮件发送（优化版）

结合 SQLAlchemy 查询，实现多收件人邮件发送。这里的关键点是利用集合自动去重，并且复用 SMTP 连接。

def send_email_to_recipients(filepath, receiver_email=None):
    """发送邮件给指定邮箱和用户关联邮箱"""
    # 获取当前用户 ID（假设通过 PassportService）
    # token, user_id = PassportService.current_user_id()
    user_id = 1  # 示例 ID

    # 收件人集合（自动去重）
    recipients = set()

    # 1. 添加直接指定的邮箱
    if receiver_email:
        recipients.add(receiver_email)

    # 2. 查询用户关联邮箱
    user = User.query.get(user_id)
    if user and user.email:
        recipients.add(user.email)

    if not recipients:
        print("无有效收件人")
        return False

    # 发送邮件（每个邮箱只发一次）
    success = True
    for email in recipients:
        if not send_email(email, "文件处理结果", "请查收附件", filepath):
            success = False
    return success

4. 优化策略

4.1 使用集合（Set）去重

在处理多个来源的收件人时，直接使用 set 可以自动过滤重复地址，无需额外逻辑。

recipients = set()
recipients.add("[email protected]")
# 再次添加相同地址不会重复
recipients.add("[email protected]") 
print(len(recipients))  # 输出 1

4.2 减少 SMTP 连接次数

频繁建立 TCP 连接会消耗资源。最佳实践是在循环外建立连接，循环内发送多条消息。

# 优化：复用 SMTP 连接
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
    server.login(sender_email, password)
    for email in recipients:
        msg['To'] = email
        server.sendmail(sender_email, email, msg.as_string())

4.3 异步发送（Celery + Redis）

对于高并发场景，同步发送会阻塞主线程。引入 Celery 将邮件任务放入队列，由 Worker 异步处理。

from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@celery.task
def async_send_email(to_email, subject, body, attachment_path=None):
    send_email(to_email, subject, body, attachment_path)

5. 完整代码示例

下面是一个整合了上述逻辑的完整示例，包含 Flask 应用结构、模型定义及发送逻辑。

# app.py
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
import os

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

# 定义 User 和 CustomerOrder 模型（略，参考上文）

def send_email_with_attachment(filepath, receiver_email=None):
    """发送邮件给指定邮箱和用户关联邮箱"""
    # 获取当前用户 ID
    user_id = 1  # 实际项目中从 Session 或 Token 获取

    # 收件人集合
    recipients = set()
    if receiver_email:
        recipients.add(receiver_email)
    
    user = User.query.get(user_id)
    if user and user.email:
        recipients.add(user.email)

    if not recipients:
        return False

    # SMTP 配置
    smtp_server = "smtp.qq.com"
    smtp_port = 465
    sender_email = "[email protected]"
    password = "your_password"

    # 创建邮件内容
    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['Subject'] = "文件处理结果"
    msg.attach(MIMEText("请查收附件", 'plain'))

    # 添加附件
    with open(filepath, "rb") as file:
        part = MIMEApplication(file.read(), Name=os.path.basename(filepath))
        part['Content-Disposition'] = f'attachment; filename="{os.path.basename(filepath)}"'
        msg.attach(part)

    # 发送邮件
    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(sender_email, password)
            for email in recipients:
                msg['To'] = email
                server.sendmail(sender_email, email, msg.as_string())
        return True
    except Exception as e:
        print(f"发送失败：{e}")
        return False

6. 总结

本文详细介绍了如何基于 Python + SQLAlchemy + SMTP 实现高效邮件发送系统，核心优化点包括：

1. 多收件人管理：利用集合自动去重，避免重复发送。
2. 附件发送支持：封装文件读取逻辑，增强稳定性。
3. 错误处理与日志：增强系统健壮性，便于排查问题。
4. 性能优化：减少 SMTP 连接次数，支持异步任务提升并发能力。

通过合理的代码设计，我们可以构建一个健壮、可扩展的邮件通知系统，适用于订单处理、文件通知等多种业务场景。