开发者

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

目录
  • 引言
  • 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 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_jsid = db.Column(db.Integer, db.ForeignKey('user.id'), nullabhttp://www.devze.comle=False)
          tracking_number = db.Column(db.javascriptString(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}")
              returaJQQzn False
      

      6. 总结

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

      1. 多收件人管理(自动去重)。
      2. 附件发送支持(文件读取优化)。
      3. 错误处理与日志(增强稳定性)。
      4. 性能优化(减少SMTP连接次数)。

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

      以上就是使用Python和SQLAlchemy实现高效的邮件发送系统的详细内容,更多关于Python SQLAlchemy邮件发送的资料请关注编程客栈(www.devze.com)其它相关文章!

      0

      上一篇:

      下一篇:

      精彩评论

      暂无评论...
      验证码 换一张
      取 消

      最新开发

      开发排行榜