SyncToAsync

方法实现

# utils/decorators.py
from typing import Callable, TypeVar
from typing_extensions import ParamSpec
from collections.abc import Coroutine
from functools import partial
import asyncio

P = ParamSpec("P")
R = TypeVar("R")

def run_sync(call: Callable[P, R]) -> Callable[P, Coroutine[None, None, R]]:
    """一个用于包装 sync function 为 async function 的装饰器
    参数:
        call: 被装饰的同步函数
    """

    @wraps(call)
    async def _wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
        loop = asyncio.get_running_loop()
        pfunc = partial(call, *args, **kwargs)
        result = await loop.run_in_executor(None, pfunc)
        return result

    return _wrapper

demo

from utils.decorators import run_sync
import asyncio, time

# 模拟处理消息耗时操作, 两条print是为了观察执行顺序
def w(i):
    print("hello", i)
    time.sleep(5)
    print("world", i)

# 模拟消息处理
async def deal(i):
    await run_sync(w)(str(i))

# 模拟消息队列
async def main():
    for i in range(10):
        asyncio.create_task(deal(i))
asyncio.run(main())

# 让程序不退出, 模拟接收消息
while 1:
    pass

输出:

hello 0
hello 1
hello 2
hello 3
hello 4
hello 5
hello 6
hello 7
hello 8
hello 9
// 过了5秒
world 0
world 1
world 2
world 3
world 4
world 5
world 6
world 7
world 8
world 9

其他方法

调用时既可以用await run_sync()()

也可以用asyncio.create_task(run_sync()())

总之, 把run_sync当作一个async函数即可

具体使用

请参阅 LegendWechatBot 的VVQuest&Meme插件