Celery中tasks.py访问模型异常：需实现多场景邮件发送

看起来你在Celery任务里访问User模型时遇到了异常，这种情况在Django+Celery的组合场景里挺常见的，我来帮你梳理几个最可能的原因和对应的解决办法：

1. Celery Worker未加载Django配置 #

Celery的worker是独立于Django主进程的，默认不会自动加载Django的应用配置，这会直接导致模型无法被识别。你需要在tasks.py的最开头添加Django初始化代码：

import os
import django
from celery import shared_task

# 设置Django配置模块的环境变量，替换成你的项目实际settings路径
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名称.settings')
# 初始化Django环境
django.setup()

# 现在再导入你的模型和邮件模块
from django.contrib.auth.models import User
from 你的邮件模块路径 import Email

2. 数据库连接过期/断开 #

Celery worker长时间运行后，可能会持有过期的数据库连接，导致查询失败。你可以在任务执行前主动关闭旧连接，确保获取新的有效连接：

@shared_task()
def send_email(sender_id=None, receiver_id=None, type=None, message=None):
    from django.db import close_old_connections
    
    # 关闭旧的数据库连接，避免使用过期无效的连接
    close_old_connections()
    
    try:
        sender = User.objects.get(id=sender_id)
        receiver = User.objects.get(id=receiver_id)
        logger.info("Starting send email")
        Email.send_email(sender, receiver, type, message)
    except User.DoesNotExist:
        logger.error(f"用户不存在：sender_id={sender_id}, receiver_id={receiver_id}")
        # 这里可以添加异常处理逻辑，比如标记任务失败、通知管理员等

3. 参数传递异常 #

如果sender_id或receiver_id传递的是None或者不存在的ID，User.objects.get()会直接抛出DoesNotExist异常。加上异常捕获不仅能避免任务崩溃，还能帮你快速定位问题根源。

4. 模型导入时机错误 #

不要在tasks.py顶部直接导入Django模型，最好是在Django初始化之后导入，或者在任务函数内部导入，这样可以避免循环导入或者未初始化的问题。

最后，记得启动Celery worker时要指定你的Django项目，比如：

celery -A 你的项目名 worker --loglevel=info

你可以先试试这些方案，看看能不能解决问题。

内容的提问来源于stack exchange，提问作者Keith Mthunzi