在事件循环已运行的情况下从同步函数运行异步代码

Question

我知道，这是一口。

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

我想使用带有异步 ORM 的 Factory-boy。唯一的问题是，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

上面的方法实际上是有效的，除非异步事件循环已经在运行。具有讽刺意味的是，该工厂通过同步测试工作，而不是通过异步测试工作。

有没有办法在异步上下文中_async_create等待？_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

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

您具体询问了事件循环已经运行的情况。在这种情况下，除了回调函数调用的代码之外，程序中运行的所有内容都是某个任务或其他任务的一部分。调用堆栈中的某个位置将是一个异步函数。如果您弄清楚如何将可等待对象传递到该级别，我认为您可以解决您的问题。