如何基于Azure实现Python邮件定时队列发送？

不需要用Kafka或RabbitMQ，Azure原生服务就能满足你的定时邮件需求，推荐两种实用方案：

方案一：Azure Queue Storage + Azure Functions（队列触发器） #

这个方案适合精确到时间点的延迟发送，利用Azure Queue的可见性延迟特性：消息存入队列后会在指定时间才变为可读取状态，随后由Azure Function自动触发发送。

步骤1：修改现有代码，将邮件任务存入队列 #

不再直接调用begin_send，而是把邮件的核心信息序列化后存入Azure Queue，设置延迟到目标发送时间：

import json
from datetime import datetime
from azure.storage.queue import QueueClient, BinaryBase64EncodePolicy

def _queue_email(self, data: Dict[str, Any], recipients: List[CustomUser], send_at: datetime) -> None:
    self._email_template.validate_data(data)
    self._initialize_message()
    relative_link = self._email_template.get_relative_link(data)
    context = self._email_template.get_template_context(data)
    context.update({"relative_link": relative_link})

    # 打包邮件任务所需的所有必要信息
    email_task = {
        "recipients": self._get_recipients(recipients),
        "subject": self._message["content"]["subject"],
        "html_content": self._render_template(context)
    }

    # 计算延迟时间（Azure Queue使用UTC时间，需统一时区）
    delay = send_at - datetime.utcnow()
    if delay.total_seconds() < 0:
        raise ValueError("发送时间不能早于当前UTC时间")

    # 连接到Azure Queue Storage
    queue_client = QueueClient.from_connection_string(
        conn_str="你的Azure Queue连接字符串",
        queue_name="email-send-tasks",
        message_encode_policy=BinaryBase64EncodePolicy()
    )

    # 发送延迟消息到队列
    queue_client.send_message(
        message_body=json.dumps(email_task),
        visibility_timeout=int(delay.total_seconds())
    )

步骤2：创建Azure Function（队列触发器） #

当队列中的消息到时间变为可见时，Function自动触发并执行邮件发送逻辑：

import json
import logging
import azure.functions as func
from azure.communication.email import EmailClient

def main(msg: func.QueueMessage) -> None:
    logging.info('开始处理队列中的邮件任务')
    email_task = json.loads(msg.get_body().decode('utf-8'))

    try:
        # 初始化ACS邮件客户端
        email_client = EmailClient.from_connection_string("你的ACS连接字符串")
        send_message = {
            "content": {
                "subject": email_task["subject"],
                "html": email_task["html_content"]
            },
            "recipients": {
                "to": email_task["recipients"]
            }
        }

        # 发送邮件
        response = email_client.begin_send(send_message)
        response.result()
        logging.info(f"邮件成功发送至: {[r['address'] for r in email_task['recipients']]}")
    except Exception as ex:
        logging.error(f"发送邮件失败: {str(ex)}")
        # 可选：添加重试逻辑，比如将消息重新入队（需限制重试次数避免死循环）

方案二：Azure Logic Apps（无代码/低代码） #

如果不想编写过多代码，可直接用Logic Apps可视化配置：

• 创建新的Logic App，添加定时触发器（设置为每天10点触发）
• 触发器触发后，添加Azure Communication Services发送邮件动作，配置邮件内容、收件人等
• 若需动态获取待发送任务，可先从Azure Storage/数据库读取邮件列表，再批量发送

注意事项 #

• Azure Queue的可见性延迟最大支持7天，若需更长周期的定时任务，建议用Logic Apps定时触发器结合Azure Table Storage存储待发送任务
• 所有连接字符串（ACS、Queue Storage）建议存入环境变量，不要硬编码
• 需添加失败重试或死信队列机制，避免邮件任务丢失

内容的提问来源于stack exchange，提问作者karol marszalek