Python asyncio场景下如何为asyncpg接口构建同步调用外观层

现有异步aiohttp Web应用通过asyncpg对接PostgreSQL数据库，无其他I/O操作。应用包含体量庞大的业务逻辑，所有逻辑仅涉及数据库I/O，无法全部改造为异步代码，因此需要实现一个非异步的业务逻辑中间层。
预期效果的伪代码示例如下：

async def handler(request):
    # 逐层调用业务代码，业务代码内仅执行SQL操作
    ...

def application_logic():
    ...
    # 直接写await会触发同步函数内的语法错误
    data = await asyncpg_conn.execute("SQL")
    ...
    # 期望实现同步调用形式
    data = asyncpg_facade.execute("SQL")
    ...

当前常见的async.run()、asyncio.run_coroutine_threadsafe()等公开方案不适用该场景，因为调用方本身已经处于异步上下文环境中，当前已有运行中的事件循环可调度asyncpg协程执行。

我们可以通过工作线程跑同步业务+主线程事件循环跑异步数据库调用的模式实现需求，全程不会阻塞主线程事件循环，改造成本极低：

1. 首先封装asyncpg的同步门面类，负责把异步调用提交到主线程事件循环执行，并在工作线程同步等待返回结果：

import asyncio
from asyncpg import Connection

class AsyncPGSyncFacade:
    def __init__(self, conn: Connection, loop: asyncio.AbstractEventLoop):
        self.conn = conn
        self.loop = loop

    def execute(self, sql: str, *args):
        future = asyncio.run_coroutine_threadsafe(
            self.conn.execute(sql, *args),
            self.loop
        )
        # 仅在工作线程等待，不会阻塞主线程
        return future.result()

    # 按照相同模式封装fetch、fetchrow、fetchval等其他需要的asyncpg接口

2. 在aiohttp的异步handler中，将同步业务逻辑放到工作线程执行，同时传入门面实例：

from aiohttp import web

async def handler(request):
    loop = asyncio.get_running_loop()
    # 从连接池获取异步数据库连接
    conn = await request.app['db_pool'].acquire()
    try:
        # 把同步业务逻辑提交到工作线程运行
        result = await asyncio.to_thread(
            application_logic,
            asyncpg_facade=AsyncPGSyncFacade(conn, loop)
        )
        return web.json_response(result)
    finally:
        # 归还连接到连接池
        await request.app['db_pool'].release(conn)

3. 现有业务逻辑无需做结构修改，直接使用传入的门面实例即可：

def application_logic(asyncpg_facade):
    # 多层业务调用直接透传asyncpg_facade即可，不需要做异步改造
    data = asyncpg_facade.execute("SELECT * FROM users WHERE id = $1", 1)
    # 其余业务逻辑完全保持原有同步写法
    return process_data(data)

该方案有两个核心优势：

• 完全符合异步框架的安全规范，不会阻塞主线程事件循环，数据库IO依然走asyncpg的纯异步实现，性能损耗极低
• 现有业务代码仅需要增加门面实例的透传逻辑，不需要修改任何业务流程，适配成本极低

附加问题1：同步函数内await为语法错误的设计逻辑 #

Python的协程是显式协作式调度设计，await只能出现在async def声明的协程函数内部，核心设计考量如下：

1. 执行语义可预判：常规同步函数的约定是"调用后会完整执行直到返回，中间不会主动让出CPU"，如果允许同步函数内使用await，就打破了这个约定，开发者无法仅通过函数声明判断调用该函数是否会触发调度，排查并发问题的成本会大幅上升。
2. 实现成本可控：如果允许跨调用栈透传异步上下文、支持任意位置await，需要解释器在所有调用栈帧维护异步上下文标记，会大幅提升解释器的实现复杂度，同时带来不必要的运行时开销，对绝大多数不需要混合调用的场景是额外负担。

附加问题2：类gevent的无侵入解决方案 #

存在类似gevent的用户态线程切换方案，核心通过greenlet实现隐式调度，不需要修改现有业务代码：
该方案的实现思路是：在同步代码调用数据库操作时，自动切出当前greenlet，把asyncpg的异步调用提交到事件循环执行，等异步调用完成后再切回对应greenlet继续执行后续逻辑。这种方案不需要修改现有业务代码，性能比线程池方案更高，也不会阻塞主线程。
但该方案也有明显缺陷：调度切换是隐式的，排查并发问题的难度更高，部分第三方同步库可能和greenlet的猴子补丁存在兼容性问题，适合业务依赖比较简单、追求低改造成本的场景。

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