使用已运行的事件循环从同步函数运行异步代码

Question

我知道，这是一张大嘴巴。

我使用的是pytest-asyncio，它提供了一个异步事件循环，用于在测试中运行异步代码。

我想使用带异步ORM的factory-boy。唯一的问题是，工厂男孩不支持任何异步。我想覆盖工厂上的_create函数(这是一个同步函数)，但是，我需要该函数来运行异步代码。它是否阻塞并不重要，它只是测试而已。

如下所示：

class AsyncModelFactory(factory.alchemy.SQLAlchemyFactory):
    class Meta:
        sqlalchemy_session = async_session
        abstract = True

    @classmethod
    def _create(cls, model_class, *args, **kwargs):
        # run_until_complete only works if there isn't already an event loop running
        return asyncio.get_event_loop().run_until_complete(
            cls._async_create(model_class, *args, **kwargs)
        )

    @classmethod
    async def _async_create(cls, model_class, *args, **kwargs):
        obj = model_class(*args, **kwargs)
        session = self._meta.sqlalchemy_session
        session.add(obj)

        await session.commit()  # needs to be awaited
        return obj

除了异步事件循环已经在运行的情况外，上述方法实际上是有效的。具有讽刺意味的是，该工厂工作于同步测试，而不是异步测试。

有没有办法在异步上下文中等待来自_create的_async_create？所有工厂男孩和工厂创建子工厂等都是同步的，所以只有当_create保持同步调用时，这才能起作用

Answer 1

当然，您不能等待普通函数中的任何内容。但是有一些普通函数和异步代码可以交互的方式。例如，一个普通的函数可以创建任务并返回可等待的对象。看看这个小程序：

import asyncio
import time

async def main():
    print("Start", time.asctime())
    awaitable1 = f1()
    awaitable2 = f2()
    await asyncio.gather(awaitable1, awaitable2)
    print("Finish", time.asctime())
    
def f1():
    return asyncio.create_task(f3())

def f2():
    return asyncio.sleep(2.0)

async def f3():
    await asyncio.sleep(1.0)
    print("Task f3", time.asctime())
    
asyncio.run(main())

结果：

Start Thu Aug 12 18:55:20 2021
Task f3 Thu Aug 12 18:55:21 2021
Finish Thu Aug 12 18:55:22 2021

这里有两个普通的函数，一个创建一个任务，另一个创建一个简单的等待值。这些对象在堆栈中向上返回到异步函数，然后等待它们。

您特别询问了事件循环已经运行的情况。在这种情况下，除了由回调函数调用的代码之外，程序中运行的所有内容都是某个任务或其他任务的一部分。调用堆栈中的某个地方将会有一个异步函数。如果你知道如何把可等待的对象传递到那个级别，我想你就能解决你的问题。