使用Python和SQLAlchemy实现高效的邮件发送系统_Python

引言

在现代web应用中，邮件通知是不可或缺的功能之一。无论是订单确认、文件处理结果通知，还是系统告警，邮件都是最常用的通信方式之一。本文将详细介绍如何基于 python、sqlalchemy 和 smtp 协议，构建一个高效、可靠的邮件发送系统。我们将从需求分析、数据库设计、代码实现到优化策略，一步步实现一个支持附件发送、多收件人管理的邮件服务。

1. 需求分析

我们的系统需要满足以下核心需求：

多收件人支持：
- 支持直接指定收件人邮箱（如 receiver_email）。
- 支持通过 user_id 查询关联的用户邮箱（存储在 user 表中）。
- 自动去重，避免重复发送。
附件发送：
- 支持发送文件附件（如csv、excel等）。
- 确保附件读取和发送的稳定性。
错误处理与日志：
- 记录邮件发送状态（成功/失败）。
- 提供详细的错误日志，便于排查问题。
性能优化：
- 避免重复构建邮件内容。
- 支持批量发送，减少smtp连接开销。

2. 数据库设计

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

2.1 user 表（存储用户信息）

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)
    # 其他字段...

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 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 = "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 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. 完整代码示例

# 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 = "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 实现高效邮件发送系统，核心优化点包括：

多收件人管理（自动去重）。
附件发送支持（文件读取优化）。
错误处理与日志（增强稳定性）。
性能优化（减少smtp连接次数）。

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

以上就是使用python和sqlalchemy实现高效的邮件发送系统的详细内容，更多关于python sqlalchemy邮件发送的资料请关注代码网其它相关文章！

使用Python和SQLAlchemy实现高效的邮件发送系统

2025年05月11日 • Python •我要评论

引言