Django新增Celery密码重置任务启动时报循环导入错误如何解决

问题根因 #

启动时触发了双向循环导入，导入链路如下：

项目启动加载app_account/models.py → 从celery_tasks导入send_password_reset_mail → celery_tasks文件头部又从models导入User模型 → 此时models还未完成初始化，直接触发导入失败。

解决方案 #

方案1（最推荐，符合Django最佳实践）：迁移信号到独立文件 #

Django的信号逻辑不建议写在models.py中，迁移后即可直接切断循环导入链路：

1. 在app_account目录下新建signals.py，将原写在models.py中的信号相关代码全部迁移进去，对应导入也一并移走：

from django.dispatch import receiver
from django_rest_passwordreset.signals import reset_password_token_created
from app_account.celery_tasks import send_password_reset_mail

# this signal for send email reset password
@receiver(reset_password_token_created)
def password_reset_token_created(sender, instance, reset_password_token, *args, **kwargs):
        email_plaintext_message = f"your token is = {reset_password_token.key}"
    
        mail_info = {
            'subject': 'Password Reset',
            'message': email_plaintext_message,
            'from_email': 'noreply@host.com',
            'recipient_list': [reset_password_token.user.email],
        }
        send_password_reset_mail.delay(mail_info)

2. 修改app_account/apps.py，在ready方法中注册信号：

from django.apps import AppConfig

class AppAccountConfig(AppConfig):
    default_auto_field = 'django.db.models.BigAutoField'
    name = 'app_account'

    def ready(self):
        # 导入信号触发注册
        import app_account.signals

3. 删除app_account/models.py中原有的信号代码，以及from app_account.celery_tasks import send_password_reset_mail这行导入语句即可。

方案2：函数内延迟导入User模型 #

如果不想调整信号位置，可将Celery任务对User的导入从文件头部移到函数内部，只有任务执行时才会导入，不会触发启动时的循环：
修改app_account/celery_tasks.py：

1. 删除文件头部的from app_account.models import User导入语句
2. 修改send_birthday_email任务，把导入放到函数内部：

@shared_task
def send_birthday_email():
    # 导入移到函数内部，执行时才加载
    from app_account.models import User
    users = User.objects.filter(is_active=True, date_of_birth__isnull=False, email__isnull=False)
    # 剩余原有逻辑保持不变

方案3：使用Celery send_task动态调用任务 #

不需要在models.py中导入Celery任务函数，直接用Celery实例的send_task方法根据任务路径调用，切断models对celery_tasks的导入依赖：
修改app_account/models.py中的信号逻辑：

1. 删除from app_account.celery_tasks import send_password_reset_mail导入语句
2. 把原有的send_password_reset_mail.delay(mail_info)替换为动态调用逻辑：

from celery import current_app

@receiver(reset_password_token_created)
def password_reset_token_created(sender, instance, reset_password_token, *args, **kwargs):
        email_plaintext_message = f"your token is = {reset_password_token.key}"
    
        mail_info = {
            'subject': 'Password Reset',
            'message': email_plaintext_message,
            'from_email': 'noreply@host.com',
            'recipient_list': [reset_password_token.user.email],
        }
        # 按任务路径动态调用，不需要提前导入任务函数
        current_app.send_task('app_account.celery_tasks.send_password_reset_mail', args=[mail_info])

内容的提问来源于stack exchange，提问作者Ali Rahmani