如何在PostgreSQL特定表新增行时触发发送模板邮件？

针对你需要在特定表新增行时自动发送包含该行数据的模板邮件的需求，以下是两种可行的实现方案，解决原生触发器直接发邮件的可靠性和功能限制问题：

方案一：异步消息队列模式（推荐，高可靠） #

这种方案将邮件发送与主数据插入解耦，触发器仅记录邮件任务，由后台服务异步处理发送，避免邮件发送失败导致事务回滚，同时提升系统性能。

实现步骤： #

1. 创建邮件任务队列表
   用于存储待发送的邮件任务，记录收件人、模板数据和状态：

CREATE TABLE email_notification_queue (
    id SERIAL PRIMARY KEY,
    recipient_email VARCHAR(255) NOT NULL,
    template_data JSONB NOT NULL, -- 存储新增行的相关字段数据
    status VARCHAR(20) DEFAULT 'pending', -- 状态：pending/sent/failed
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

2. 编写触发器函数
   当目标表插入新行时，将邮件任务写入队列表。假设目标表为user_registrations，包含id、email、username、register_time字段：

CREATE OR REPLACE FUNCTION queue_email_notification()
RETURNS TRIGGER AS $$
BEGIN
    INSERT INTO email_notification_queue (recipient_email, template_data)
    VALUES (
        NEW.email,
        jsonb_build_object(
            'user_id', NEW.id,
            'username', NEW.username,
            'register_time', NEW.register_time
        )
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

3. 绑定触发器到目标表
   设置在user_registrations表插入后触发上述函数：

CREATE TRIGGER after_user_reg_insert
AFTER INSERT ON user_registrations
FOR EACH ROW EXECUTE FUNCTION queue_email_notification();

4. 编写后台邮件处理服务
   用Python编写定时常驻服务，读取队列中的待发送任务，调用邮件服务发送，并更新任务状态：

import psycopg2
import smtplib
from email.mime.text import MIMEText
from datetime import datetime
import time

# 数据库配置
DB_SETTINGS = {
    "dbname": "your_database",
    "user": "your_db_user",
    "password": "your_db_password",
    "host": "localhost"
}

# SMTP邮件配置
SMTP_SETTINGS = {
    "server": "smtp.example.com",
    "port": 587,
    "sender": "notifications@example.com",
    "password": "your_smtp_password"
}

def send_registration_email(recipient, user_data):
    """构建并发送注册通知邮件"""
    subject = "欢迎加入我们的平台"
    body = f"""
    尊敬的{user_data['username']}：
    您已成功完成注册，您的用户ID为{user_data['user_id']}，注册时间：{user_data['register_time']}。
    感谢您的支持！
    """
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = SMTP_SETTINGS["sender"]
    msg["To"] = recipient

    try:
        with smtplib.SMTP(SMTP_SETTINGS["server"], SMTP_SETTINGS["port"]) as server:
            server.starttls()
            server.login(SMTP_SETTINGS["sender"], SMTP_SETTINGS["password"])
            server.send_message(msg)
        return True
    except Exception as e:
        print(f"邮件发送失败：{str(e)}")
        return False

def process_email_queue():
    """处理队列中的待发送邮件任务"""
    conn = psycopg2.connect(**DB_SETTINGS)
    cur = conn.cursor()

    # 锁定待处理任务，避免多进程重复处理
    cur.execute("""
        UPDATE email_notification_queue
        SET status = 'processing', updated_at = %s
        WHERE id IN (
            SELECT id FROM email_notification_queue
            WHERE status = 'pending'
            LIMIT 10
            FOR UPDATE SKIP LOCKED
        )
        RETURNING id, recipient_email, template_data;
    """, (datetime.now(),))

    tasks = cur.fetchall()
    for task_id, recipient, user_data in tasks:
        success = send_registration_email(recipient, user_data)
        if success:
            cur.execute("""
                UPDATE email_notification_queue
                SET status = 'sent', updated_at = %s
                WHERE id = %s;
            """, (datetime.now(), task_id))
        else:
            cur.execute("""
                UPDATE email_notification_queue
                SET status = 'failed', updated_at = %s
                WHERE id = %s;
            """, (datetime.now(), task_id))
        conn.commit()

    cur.close()
    conn.close()

if __name__ == "__main__":
    # 循环监听队列，每分钟处理一次
    while True:
        process_email_queue()
        time.sleep(60)

方案二：触发器直接调用外部脚本（简单场景适用） #

如果你的系统并发量低，且可以接受邮件发送失败导致插入回滚的风险，可以用触发器直接调用外部脚本发送邮件。

实现步骤： #

1. 启用plsh扩展
   允许PostgreSQL执行shell命令：

CREATE EXTENSION IF NOT EXISTS plsh;

2. 编写触发器函数
   调用外部Python脚本传递邮件信息：

CREATE OR REPLACE FUNCTION send_email_on_insert()
RETURNS TRIGGER AS $$
BEGIN
    -- 调用外部脚本，传递收件人和用户数据
    PERFORM plsh_exec(
        '/usr/bin/python3 /path/to/your/send_email.py ' ||
        quote_literal(NEW.email) || ' ' ||
        quote_literal(jsonb_build_object(
            'user_id', NEW.id,
            'username', NEW.username
        )::text)
    );
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

3. 绑定触发器

CREATE TRIGGER after_user_reg_insert
AFTER INSERT ON user_registrations
FOR EACH ROW EXECUTE FUNCTION send_email_on_insert();

4. 编写外部发送脚本

import sys
import smtplib
from email.mime.text import MIMEText
import json

def send_email(recipient, user_data):
    subject = "注册成功通知"
    body = f"您好{user_data['username']}，您的用户ID是{user_data['user_id']}，欢迎加入！"
    msg = MIMEText(body, "plain", "utf-8")
    msg["Subject"] = subject
    msg["From"] = "notifications@example.com"
    msg["To"] = recipient

    with smtplib.SMTP("smtp.example.com", 587) as server:
        server.starttls()
        server.login("notifications@example.com", "your_smtp_password")
        server.send_message(msg)

if __name__ == "__main__":
    recipient = sys.argv[1]
    user_data = json.loads(sys.argv[2])
    send_email(recipient, user_data)

关键注意事项 #

• 方案一优势：异步处理不阻塞主事务，邮件失败可重试，适合高并发、数据一致性要求高的场景。
• 方案二限制：同步执行会增加插入耗时，邮件发送失败会导致插入回滚，仅适合低并发场景。
• 推荐使用专业邮件服务商（如SendGrid、Mailgun）代替自建SMTP，提升邮件送达率。
• 确保后台服务或PostgreSQL进程有足够的权限访问邮件服务和数据库。

内容的提问来源于stack exchange，提问作者Martin Iglesias