使用Python和SQLAlchemy實(shí)現(xiàn)高效的郵件發(fā)送系統(tǒng)

更新時(shí)間：2025年05月11日 09:41:12 作者：碼農(nóng)阿豪@新空間

在現(xiàn)代Web應(yīng)用中,郵件通知是不可或缺的功能之一,無論是訂單確認(rèn)、文件處理結(jié)果通知,還是系統(tǒng)告警,郵件都是最常用的通信方式之一,本文將詳細(xì)介紹如何基于 Python、SQLAlchemy 和 SMTP 協(xié)議,構(gòu)建一個(gè)高效、可靠的郵件發(fā)送系統(tǒng),需要的朋友可以參考下

引言

在現(xiàn)代Web應(yīng)用中，郵件通知是不可或缺的功能之一。無論是訂單確認(rèn)、文件處理結(jié)果通知，還是系統(tǒng)告警，郵件都是最常用的通信方式之一。本文將詳細(xì)介紹如何基于 Python、SQLAlchemy 和 SMTP 協(xié)議，構(gòu)建一個(gè)高效、可靠的郵件發(fā)送系統(tǒng)。我們將從需求分析、數(shù)據(jù)庫(kù)設(shè)計(jì)、代碼實(shí)現(xiàn)到優(yōu)化策略，一步步實(shí)現(xiàn)一個(gè)支持附件發(fā)送、多收件人管理的郵件服務(wù)。

1. 需求分析

我們的系統(tǒng)需要滿足以下核心需求：

多收件人支持：
- 支持直接指定收件人郵箱（如 receiver_email）。
- 支持通過 user_id 查詢關(guān)聯(lián)的用戶郵箱（存儲(chǔ)在 User 表中）。
- 自動(dòng)去重，避免重復(fù)發(fā)送。
附件發(fā)送：
- 支持發(fā)送文件附件（如CSV、Excel等）。
- 確保附件讀取和發(fā)送的穩(wěn)定性。
錯(cuò)誤處理與日志：
- 記錄郵件發(fā)送狀態(tài)（成功/失敗）。
- 提供詳細(xì)的錯(cuò)誤日志，便于排查問題。
性能優(yōu)化：
- 避免重復(fù)構(gòu)建郵件內(nèi)容。
- 支持批量發(fā)送，減少SMTP連接開銷。

2. 數(shù)據(jù)庫(kù)設(shè)計(jì)

郵件發(fā)送系統(tǒng)通常需要關(guān)聯(lián)用戶數(shù)據(jù)，因此我們使用 SQLAlchemy 定義數(shù)據(jù)模型：

2.1 User 表（存儲(chǔ)用戶信息）

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 表（關(guān)聯(lián)用戶訂單）

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表的關(guān)系
    user = db.relationship('User', backref='orders')

3. 郵件發(fā)送核心實(shí)現(xiàn)

3.1 基礎(chǔ)郵件發(fā)送（SMTP）

我們使用Python的 smtplib 和 email 庫(kù)實(shí)現(xiàn)郵件發(fā)送：

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):
    """發(fā)送郵件（支持附件）"""
    # 郵件服務(wù)器配置
    smtp_server = "smtp.qq.com"
    smtp_port = 465
    sender_email = "your_email@qq.com"
    password = "your_smtp_password"  # 建議使用環(huán)境變量
    
    # 創(chuàng)建郵件對(duì)象
    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)
    
    # 發(fā)送郵件
    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"郵件發(fā)送失敗: {e}")
        return False

3.2 多收件人郵件發(fā)送（優(yōu)化版）

結(jié)合SQLAlchemy查詢，實(shí)現(xiàn)多收件人郵件發(fā)送：

def send_email_to_recipients(filepath, receiver_email=None):
    """發(fā)送郵件給指定郵箱和用戶關(guān)聯(lián)郵箱"""
    # 獲取當(dāng)前用戶ID（假設(shè)通過PassportService）
    token, user_id = PassportService.current_user_id()
    
    # 收件人集合（自動(dòng)去重）
    recipients = set()
    
    # 1. 添加直接指定的郵箱
    if receiver_email:
        recipients.add(receiver_email)
    
    # 2. 查詢用戶關(guān)聯(lián)郵箱
    user = User.query.get(user_id)
    if user and user.email:
        recipients.add(user.email)
    
    if not recipients:
        print("無有效收件人")
        return False
    
    # 發(fā)送郵件（每個(gè)郵箱只發(fā)一次）
    success = True
    for email in recipients:
        if not send_email(email, "文件處理結(jié)果", "請(qǐng)查收附件", filepath):
            success = False
    
    return success

4. 優(yōu)化策略

4.1 使用集合（Set）去重

recipients = set()
recipients.add("user1@example.com")  # 自動(dòng)去重

4.2 減少SMTP連接次數(shù)

# 優(yōu)化：復(fù)用SMTP連接
with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
    server.login(sender_email, password)
    for email in recipients:
        server.sendmail(...)

4.3 異步發(fā)送（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):
    """發(fā)送郵件給指定郵箱和用戶關(guān)聯(lián)郵箱"""
    # 獲取當(dāng)前用戶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"
    
    # 創(chuàng)建郵件內(nèi)容
    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['Subject'] = "文件處理結(jié)果"
    msg.attach(MIMEText("請(qǐng)查收附件", '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)
    
    # 發(fā)送郵件
    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"發(fā)送失敗: {e}")
        return False