Implement efficient mail sending system using Python and SQLAlchemy

introduction

In modern web applications, email notification is one of the indispensable features. Whether it is order confirmation, file processing result notification, or system alarm, email is one of the most commonly used communication methods. This article will introduce in detail how to build an efficient and reliable mail sending system based on Python, SQLAlchemy and SMTP protocols. We will implement a mail service that supports attachment sending and multi-recipient management step by step from requirements analysis, database design, code implementation to optimization strategy.

1. Requirements Analysis

Our system needs to meet the following core needs:

Multi-recipient support:
- Supports direct specifying the recipient's email address (such asreceiver_email）。
- Supporteduser_idQuery the associated user mailbox (stored inUserin the table).
- Automatically deduplicate to avoid repeated sending.
Attachment sent:
- Support sending file attachments (such as CSV, Excel, etc.).
- Ensures stability of attachment reading and sending.
Error handling and logging:
- Record the sending status of the email (success/failure).
- Provide detailed error logs to facilitate troubleshooting.
Performance optimization:
- Avoid repeated builds of email content.
- Supports batch sending to reduce SMTP connection overhead.

2. Database design

Email sending systems usually require associated user data, so we use SQLAlchemy to define the data model:

2.1 User table (storing user information)

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User():
    __tablename__ = 'user'
    
    id = (, primary_key=True)
    email = ((120), nullable=False, unique=True)
    username = ((80), nullable=False)
    #Other fields...

2.2 CustomerOrder table (associated user order)

class CustomerOrder():
    __tablename__ = 'customer_order'
    
    id = (, primary_key=True)
    user_id = (, (''), nullable=False)
    tracking_number = ((50), nullable=False)
    order_number = ((50), nullable=False)
    #Other fields...    
    # Define the relationship with the User table    user = ('User', backref='orders')

3. Core implementation of email sending

3.1 Basic mail sending (SMTP)

We use PythonsmtplibandemailThe library implements email sending:

import smtplib
from  import MIMEMultipart
from  import MIMEText
from  import MIMEApplication
import os

def send_email(to_email, subject, body, attachment_path=None):
    """Send email (support attachments)"""
    # Email Server Configuration    smtp_server = ""
    smtp_port = 465
    sender_email = "your_email@"
    password = "your_smtp_password"  # It is recommended to use environment variables    
    # Create a mail object    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['To'] = to_email
    msg['Subject'] = subject
    
    # Add text    (MIMEText(body, 'plain'))
    
    # Add attachment (if any)    if attachment_path:
        with open(attachment_path, "rb") as file:
            part = MIMEApplication((), Name=(attachment_path))
            part['Content-Disposition'] = f'attachment; filename="{(attachment_path)}"'
            (part)
    
    # Send email    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            (sender_email, password)
            (sender_email, to_email, msg.as_string())
        return True
    except Exception as e:
        print(f"Email sending failed: {e}")
        return False

3.2 Multi-recipient email sending (optimized version)

Combined with SQLAlchemy query, multi-recipient email sending:

def send_email_to_recipients(filepath, receiver_email=None):
    """Send emails to the specified mailbox and user-related mailbox"""
    # Get the current user ID (assuming it passes PassportService)    token, user_id = PassportService.current_user_id()
    
    # Recipient collection (automatic deduplication)    recipients = set()
    
    # 1. Add the mailbox you specified directly    if receiver_email:
        (receiver_email)
    
    # 2. Query the user's associated mailbox    user = (user_id)
    if user and :
        ()
    
    if not recipients:
        print("No valid recipient")
        return False
    
    # Send emails (only send once per email)    success = True
    for email in recipients:
        if not send_email(email, "File Processing Results", "Please check the attachment", filepath):
            success = False
    
    return success

4. Optimization strategy

4.1 Use Sets to deduplicate

recipients = set()
("user1@")  # Automatically dereload

4.2 Reduce the number of SMTP connections

# Optimization: Multiplexing SMTP connectionswith smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
    (sender_email, password)
    for email in recipients:
        (...)

4.3 Asynchronous sending (Celery + Redis)

from celery import Celery

celery = Celery('tasks', broker='redis://localhost:6379/0')

@
def async_send_email(to_email, subject, body, attachment_path=None):
    send_email(to_email, subject, body, attachment_path)

5. Complete code example

# 
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
import smtplib
from  import MIMEMultipart
from  import MIMEText
from  import MIMEApplication
import os

app = Flask(__name__)
['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///'
db = SQLAlchemy(app)

# Define User and CustomerOrder models (omitted)
def send_email_with_attachment(filepath, receiver_email=None):
    """Send emails to the specified mailbox and user-related mailbox"""
    # Get the current user ID    token, user_id = PassportService.current_user_id()
    
    # Recipient Collection    recipients = set()
    if receiver_email:
        (receiver_email)
    
    user = (user_id)
    if user and :
        ()
    
    if not recipients:
        return False
    
    # SMTP configuration    smtp_server = ""
    smtp_port = 465
    sender_email = "your_email@"
    password = "your_password"
    
    # Create email content    msg = MIMEMultipart()
    msg['From'] = sender_email
    msg['Subject'] = "File Processing Results"
    (MIMEText("Please check the attachment", 'plain'))
    
    # Add attachment    with open(filepath, "rb") as file:
        part = MIMEApplication((), Name=(filepath))
        part['Content-Disposition'] = f'attachment; filename="{(filepath)}"'
        (part)
    
    # Send email    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            (sender_email, password)
            for email in recipients:
                msg['To'] = email
                (sender_email, email, msg.as_string())
        return True
    except Exception as e:
        print(f"Send failed: {e}")
        return False

6. Summary

This article introduces in detail how to implement an efficient mail sending system based on Python + SQLAlchemy + SMTP. The core optimization points include:

Multi-recipient management (automatic deduplication).
Attachment sending support (file read optimization).
Error handling and logging (enhanced stability).
Performance optimization (reduce the number of SMTP connections).

Through reasonable code design, we can build a robust and scalable email notification system suitable for order processing, file notification and other scenarios.

The above is the detailed content of using Python and SQLAlchemy to implement an efficient email sending system. For more information about Python SQLAlchemy email sending, please follow my other related articles!