Celery任务调用Django异步ORM报错的原因及解决方法

问题场景 #

使用Django 4.2.6（支持异步ORM方法如aget_or_create、aget）编写了以下代码：

BookmakerA类实现 #

class BookmakerA:
    def __init__(self) -> None:
        self.bookmaker = None 
        
    async def _init(self):
        self.bookmaker, _ = await Bookmaker.objects.aget_or_create(name="BookmakerA", defaults={"name": "BookmakerA"})

Celery同步任务调用 #

@shared_task
def get_bookmaker_matches(): 
    start = time.time()

    bookmakera = BookmakerA() 
    
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    
    loop.run_until_complete(bookmakera._init())

运行后触发错误：

django.core.exceptions.SynchronousOnlyOperation: You cannot call this from an async context - use a thread or sync_to_async.

即使改用以下写法，仍出现相同错误：

await sync_to_async(Bookmaker.objects.get_or_create, thread_sensitive=True)(name="BookmakerA", defaults={"name": "BookmakerA"})

错误原因 #

1. 上下文不匹配：Celery同步任务中手动创建的asyncio事件循环，未初始化Django的异步运行上下文。Django的异步ORM方法依赖自身提供的异步环境（如ASGI请求上下文、Django初始化的async循环）来管理数据库连接和状态，普通asyncio循环无法满足要求。
2. sync_to_async使用逻辑错误：在已有的async上下文里用sync_to_async包裹同步ORM方法，会导致同步代码尝试在异步线程环境中运行，但此时Django的同步上下文未正确初始化，引发冲突。

解决方法 #

方案1：使用Celery异步任务（推荐，Celery 5.0+支持） #

将Celery任务改为异步类型，由Celery自动处理事件循环和上下文：

@shared_task(asyncio=True)
async def get_bookmaker_matches(): 
    start = time.time()
    bookmakera = BookmakerA() 
    await bookmakera._init()

方案2：用async_to_sync转换异步方法（适合保留同步任务结构） #

利用Django的async_to_sync工具，将异步的_init方法转换为同步调用，无需手动管理事件循环：

from asgiref.sync import async_to_sync

@shared_task
def get_bookmaker_matches(): 
    start = time.time()

    bookmakera = BookmakerA() 
    async_to_sync(bookmakera._init)()

方案3：手动初始化Django异步上下文（不推荐，仅作参考） #

如果必须手动创建事件循环，需先初始化Django的ASGI上下文：

@shared_task
def get_bookmaker_matches(): 
    start = time.time()

    import asyncio
    from django.core.asgi import get_asgi_application
    import django

    # 初始化Django环境
    django.setup()
    asgi_app = get_asgi_application()

    async def run():
        # 进入ASGI上下文
        scope = {'type': 'http'}
        async with asgi_app(scope, lambda r, s: None, lambda s: None):
            bookmakera = BookmakerA() 
            await bookmakera._init()

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(run())

内容的提问来源于stack exchange，提问作者Sander Bakker