如何使用 Python asyncio 在 asyncpg API 上建立同步外观？

Question

想象一下一个异步aiohttpWeb 应用程序，它由通过连接的 Postgresql 数据库支持asyncpg，并且不执行其他 I/O。我怎样才能有一个托管应用程序逻辑的中间层，而不是异步的？（我知道我可以简单地使所有内容异步 - 但想象我的应用程序具有大量应用程序逻辑，仅受数据库 I/O 约束，并且我无法触及其中的所有内容）。

\n

伪代码：

\n

async def handler(request):\n    # call into layers over layers of application code, that simply emits SQL\n    ...\n\ndef application_logic():\n    ...\n    # This doesn\'t work, obviously, as await is a syntax\n    # error inside synchronous code.\n    data = await asyncpg_conn.execute("SQL")\n    ...\n    # What I want is this:\n    data = asyncpg_facade.execute("SQL")\n    ...\n

\n

如何asyncpg构建同步 fa\xc3\xa7ade 来允许应用程序逻辑进行数据库调用？async.run()在这种情况下，诸如 using或等浮动的配方asyncio.run_coroutine_threadsafe()不起作用，因为我们来自已经异步的上下文。我认为这不可能是不可能的，因为已经有一个事件循环原则上可以运行协asyncpg程。

\n

await额外问题：使inside 同步成为语法错误的设计原理是什么？await允许来自协程的任何上下文不是非常有用，这样我们就可以通过简单的方法将应用程序分解为功能构建块吗？

\n

编辑额外的奖励：除了保罗非常好的答案（留在“安全区域”内）之外，我对避免阻塞主线程（导致更多 gevent 的东西）的解决方案感兴趣。另请参阅我对保罗的回答的评论......

\n

Answer 1

您需要创建一个运行异步代码的辅助线程。您可以使用辅助线程自己的事件循环来初始化辅助线程，该事件循环将永远运行。通过调用 run_coroutine_threadsafe() 并在返回的对象上调用 result() 来执行每个异步函数。这是一个并发的实例。futures.Future，它的 result() 方法不会返回，直到协程的结果从辅助线程准备好。

然后，您的主线程实际上会调用每个异步函数，就好像它是同步函数一样。在每个函数调用完成之前，主线程不会继续执行。顺便说一句，您的同步函数是否实际上在事件循环上下文中运行并不重要。

当然，对 result() 的调用将阻塞主线程的事件循环。如果您想获得从同步代码运行异步函数的效果，这是不可避免的。

不用说，这是一件丑陋的事情，并且暗示着错误的程序结构。但您正在尝试转换遗留程序，它可能会有所帮助。

import asyncio
import threading
from datetime import datetime

def main():
    def thr(loop):
        asyncio.set_event_loop(loop)
        loop.run_forever()
    
    loop = asyncio.new_event_loop()
    t = threading.Thread(target=thr, args=(loop, ), daemon=True)
    t.start()

    print("Hello", datetime.now())
    t1 = asyncio.run_coroutine_threadsafe(f1(1.0), loop).result()
    t2 = asyncio.run_coroutine_threadsafe(f1(2.0), loop).result()
    print(t1, t2)
 

if __name__ == "__main__":
    main()

>>> Hello 2021-10-26 20:37:00.454577
>>> Hello 1.0 2021-10-26 20:37:01.464127
>>> Hello 2.0 2021-10-26 20:37:03.468691
>>> 1.0 2.0