Celery任务调用Django ORM偶发SynchronousOnlyOperation错误如何解决

1. 报错偶发的原因 #

Django 3.1+ 新增了异步上下文安全检测，当检测到当前执行环境存在活跃的asyncio事件循环时，直接调用同步ORM操作就会抛出SynchronousOnlyOperation错误。
你使用Celery的gevent启动模式，gevent会对Python标准库做猴子补丁，模拟实现事件循环，部分场景下会被Django的检测逻辑误判为异步上下文。这种误判不是每次任务执行都会触发，受worker当前运行状态、之前执行过的任务类型影响，所以报错表现为偶发。

2. 可行解决方案 #

首先解释你使用sync_to_async时报错的原因：sync_to_async包裹同步逻辑后返回的是协程对象，需要在异步函数中通过await执行才能拿到实际返回结果，你直接在同步Celery任务中调用，没有等待协程执行，直接使用返回的协程对象做迭代等操作，就会触发TypeError: 'coroutine' object is not iterable。

你可以根据业务场景选择以下任意一种方案解决：

• 方案一：关闭Django异步上下文安全检测
  直接在Django的settings.py文件头部添加以下代码即可，该配置是Django官方提供的适配方案，适合确认异步上下文下调用同步ORM不存在并发安全问题的场景（gevent模式下ORM操作默认是同步阻塞的，不存在安全问题）：

  import os
  os.environ["DJANGO_ALLOW_ASYNC_UNSAFE"] = "true"
  

• 方案二：更换Celery worker的并发模式
  将启动命令中的gevent模式替换为prefork模式，修改后启动命令如下，prefork模式下不存在事件循环检测误判的问题：

  celery -A Tasks worker -P prefork -c 10 -l INFO -E
  

• 方案三：适配同步任务下的协程执行
  如果需要保留gevent模式和Django的安全检测，可以在调用sync_to_async包裹的ORM逻辑时，通过asyncio.run执行协程拿到结果，示例如下：

  import asyncio
  from asgiref.sync import sync_to_async
  
  def your_celery_task():
      # 假设要执行User.objects.all()查询
      queryset = asyncio.run(sync_to_async(User.objects.all)())
      # 后续正常使用queryset即可
  

内容的提问来源于stack exchange，提问作者Ruben Rehn