Python
高效邮件发送系统设计与实现：基于 Python 和 SQLAlchemy 的实践
基于 Python、SQLAlchemy 和 SMTP 协议构建高效邮件发送系统。涵盖需求分析、数据库设计、核心代码实现及优化策略。支持多收件人自动去重、附件发送、错误日志记录。通过集合去重、复用 SMTP 连接及异步任务队列（Celery）提升性能。适用于订单确认、系统告警等场景。
基于 Python、SQLAlchemy 和 SMTP 协议构建高效邮件发送系统。涵盖需求分析、数据库设计、核心代码实现及优化策略。支持多收件人自动去重、附件发送、错误日志记录。通过集合去重、复用 SMTP 连接及异步任务队列（Celery）提升性能。适用于订单确认、系统告警等场景。
在现代 Web 应用中，邮件通知是不可或缺的功能之一。无论是订单确认、文件处理结果通知，还是系统告警，邮件都是最常用的通信方式之一。本文将详细介绍如何基于 Python、SQLAlchemy 和 SMTP 协议，构建一个高效、可靠的邮件发送系统。我们将从需求分析、数据库设计、代码实现到优化策略，一步步实现一个支持附件发送、多收件人管理的邮件服务。
我们的系统需要满足以下核心需求：
receiver_email）。user_id 查询关联的用户邮箱（存储在 User 表中）。邮件发送系统通常需要关联用户数据，因此我们使用 SQLAlchemy 定义数据模型：
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class User(db.Model):
__tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True)
email = db.Column(db.String(120), nullable=False, unique=True)
username = db.Column(db.String(80), nullable=False)
# 其他字段...
class CustomerOrder(db.Model):
__tablename__ = 'customer_order'
id = db.Column(db.Integer, primary_key=True)
user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
tracking_number = db.Column(db.String(50), nullable=False)
order_number = db.Column(db.String(50), nullable=False)
# 其他字段...
# 定义与 User 表的关系
user = db.relationship('User', backref='orders')
我们使用 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
结合 SQLAlchemy 查询，实现多收件人邮件发送：
def send_email_to_recipients(filepath, receiver_email=None):
"""发送邮件给指定邮箱和用户关联邮箱"""
# 获取当前用户 ID（假设通过 PassportService）
token, user_id = PassportService.current_user_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
recipients = set()
recipients.add("[email protected]") # 自动去重
# 优化：复用 SMTP 连接
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
server.login(sender_email, password)
for email in recipients:
server.sendmail(...)
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)
# 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
token, user_id = PassportService.current_user_id()
# 收件人集合
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
本文详细介绍了如何基于 Python + SQLAlchemy + SMTP 实现高效邮件发送系统，核心优化点包括：
通过合理的代码设计，我们可以构建一个健壮、可扩展的邮件通知系统，适用于订单处理、文件通知等场景。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
解析常见 curl 参数并生成 fetch、axios、PHP curl 或 Python requests 示例代码。 在线工具，curl 转代码在线工具，online
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML转Markdown 互为补充。 在线工具，Markdown转HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML转Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online