Celery周期任务读取数据库数据：每日日期无法动态更新问题

问题核心 #

你的代码在应用启动时就计算了today并一次性创建了所有定时任务，但这些任务不会每天重新查询当日生日用户，导致后续日期无法给正确的用户发送邮件。

修复方案 #

需要把「每日获取当日生日用户」和「发送邮件」拆成两个独立步骤：让Celery每天固定时间先执行查询逻辑，再为每个用户触发邮件发送任务，而非启动时就把所有任务固定下来。

步骤1：封装独立的邮件发送任务 #

先把发送邮件的逻辑单独做成一个Celery任务，方便后续调用：

@celery.task
def send_birthday_email(email, first_name, last_name):
    # 替换为你的实际邮件发送逻辑（比如用smtplib或Flask-Mail）
    print(f"发送生日问候给 {first_name} {last_name} <{email}>")

步骤2：定义每日执行的查询+触发任务 #

创建一个每天定时运行的任务，负责查询当日生日用户，再逐个调用邮件发送任务：

@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每天12:20执行一次查询并触发邮件
    sender.add_periodic_task(
        crontab(hour=12, minute=20),
        fetch_today_birthdays_and_send.s(),
    )

@celery.task
def fetch_today_birthdays_and_send():
    # 每次执行时才计算当日日期，保证是最新的
    today = datetime.now().strftime("%d/%m")
    emails = DB.get_email_today(today)
    for row in emails:
        send_birthday_email.delay(row['email'], row['first_name'], row['last_name'])

完整代码示例 #

from flask import Flask
from celery.schedules import crontab
from celery import Celery
from datetime import datetime

app = Flask(__name__)
celery = Celery(app.name)  # 补充Celery初始化，原代码可能遗漏了这部分

@celery.task
def send_birthday_email(email, first_name, last_name):
    # 替换为你的实际邮件发送逻辑
    print(f"发送生日问候邮件至：{first_name} {last_name} <{email}>")

@celery.task
def fetch_today_birthdays_and_send():
    today = datetime.now().strftime("%d/%m")
    emails = DB.get_email_today(today)
    for row in emails:
        send_birthday_email.delay(row['email'], row['first_name'], row['last_name'])

@celery.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    # 每天12点20分执行查询并发送邮件
    sender.add_periodic_task(
        crontab(hour=12, minute=20),
        fetch_today_birthdays_and_send.s(),
    )

关键改进点 #

• 延迟日期计算：把today的计算移到每日执行的任务内部，确保每次运行时都是当天的日期。
• 动态生成任务：每天重新查询当日用户，再逐个触发邮件发送，避免启动时就固定所有任务的问题。
• 逻辑解耦：查询和发送拆分为独立任务，代码更清晰，也方便单独调试某一部分功能。

内容的提问来源于stack exchange，提问作者Alex K