目录

• 引言
• 1. 需求分析
• 2. 数据库设计
• 2.1 User 表（存储用户信息）
• 2.2 CustomerOrder 表（关联用户订单）
• 3. 邮件发送核心实现
• 3.1 基础邮件发送（SMTP）
• 3.2 多收件人邮件发送（优化版）
• 4. 优化策略
• 4.1 使用集合（Set）去重
• 4.2 减少SMTP连接次数
• 4.3 异步发送（Celery + Redis）
• 5. 完整代码示例
• 6. 拓展资料

引言

在现代Web应用中，邮件通知是不可或缺的功能其中一个。无论是订单确认、文件处理结局通知，还是体系告警，邮件都是最常用的通信方式其中一个。这篇文章小编将详细介绍怎样基于 Python、SQLAlchemy 和 SMTP 协议，构建一个高效、可靠的邮件发送体系。我们将从需求分析、数据库设计、代码实现到优化策略，一步步实现一个支持附件发送、多收件人管理的邮件服务。

1. 需求分析

我们的体系需要满足下面内容核心需求：

1. 多收件人支持：

   • 支持直接指定收件人邮箱（如receiver_email）。
   • 支持通过user_id查询关联的用户邮箱（存储在User表中）。
   • 自动去重，避免重复发送。

2. 附件发送：

   • 支持发送文件附件（如CSV、Excel等）。
   • 确保附件读取和发送的稳定性。

3. 错误处理与日志：

   • 记录邮件发送情形（成功/失败）。
   • 提供详细的错误日志，便于排查难题。

4. 性能优化：

   • 避免重复构建邮件内容。
   • 支持批量发送，减少SMTP连接开销。

2. 数据库设计

邮件发送体系通常需要关联用户数据，因此我们使用 SQLAlchemy 定义数据模型：

2.1 User 表（存储用户信息）

from flask_sqlalchemy import SQLAlchemydb = 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) 其他字段…

2.2 CustomerOrder 表（关联用户订单）

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’)

3. 邮件发送核心实现

3.1 基础邮件发送（SMTP）

我们使用Python的smtplib和email库实现邮件发送：

import smtplibfrom email.mime.multipart import MIMEMultipartfrom email.mime.text import MIMETextfrom email.mime.application import MIMEApplicationimport osdef send_email(to_email, subject, body, attachment_path=None): “””发送邮件（支持附件）””” 邮件服务器配置 smtp_server = “smtp.qq.com” smtp_port = 465 sender_email = “your_email@qq.com” 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查询，实现多收件人邮件发送：

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

4. 优化策略

4.1 使用集合（Set）去重

recipients = set()recipients.add(“user1@example.com”) 自动去重

4.2 减少SMTP连接次数

优化：复用SMTP连接with smtplib.SMTP_SSL(smtp_server, smtp_port) as server: server.login(sender_email, password) for email in recipients: server.sendmail(…)

4.3 异步发送（Celery + Redis）

from celery import Celerycelery = Celery(‘tasks’, broker=’redis://localhost:6379/0′)@celery.taskdef async_send_email(to_email, subject, body, attachment_path=None): send_email(to_email, subject, body, attachment_path)

5. 完整代码示例

app.pyfrom flask import Flaskfrom flask_sqlalchemy import SQLAlchemyimport smtplibfrom email.mime.multipart import MIMEMultipartfrom email.mime.text import MIMETextfrom email.mime.application import MIMEApplicationimport osapp = 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 = “your_email@qq.com” 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连接次数）。

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

以上就是使用Python和SQLAlchemy实现高效的邮件发送体系的详细内容，更多关于Python SQLAlchemy邮件发送的资料请关注风君子博客其它相关文章！

无论兄弟们可能感兴趣的文章:

• Python使用SQLAlchemy连接ClickHouse数据库
• pythonSQLAlchemy数据库连接池的实现
• Python怎样使用sqlalchemy实现动态sql
• Python使用SQLAlchemy进行复杂查询的操作代码
• Python中SQLAlchemy库的使用技巧分析