是否可以让装饰器在 asyncio 执行器中运行阻塞函数？

Question

我可能需要帮助更好地表达这个问题。我正在通过 python3.7 和一个类（称为Worker()）编写一个异步 api 接口。Worker有一些我想使用运行的阻塞方法loop.run_in_executor()。

我想构建一个装饰器，我可以async在所有非方法之上添加Worker，但我不断遇到问题。

有人告诉我我需要await wraps()在下面的装饰器中：

def run_method_in_executor(func, *, loop=None):
    async def wraps(*args):
        _loop = loop if loop is not None else asyncio.get_event_loop()
        return await _loop.run_in_executor(executor=None, func=func, *args)
    return wraps

这会抛出： RuntimeWarning: coroutine 'run_method_in_executor.<locals>.wraps' was never awaited

我不知道如何正确地执行，await wraps()因为包含函数和修饰函数不是异步的。不确定这是由于误解asyncio还是装饰器的误解。

任何帮助（或帮助澄清）将不胜感激！

Answer 1

这是 Python 3.6+ 的完整示例，它不使用 3.8 弃用的接口。返回 的值loop.run_in_executor可以有效地将包装函数转换为在线程中执行的可等待await函数，以便您可以完成它。

#!/usr/bin/env python3

import asyncio
import functools
import time


def run_in_executor(_func):
    @functools.wraps(_func)
    def wrapped(*args, **kwargs):
        loop = asyncio.get_event_loop()
        func = functools.partial(_func, *args, **kwargs)
        return loop.run_in_executor(executor=None, func=func)
    return wrapped


@run_in_executor
def say(text=None):
    """Block, then print."""
    time.sleep(1.0)
    print(f'say {text} at {time.monotonic():.3f}')


async def main():
    print(f'beginning at {time.monotonic():.3f}')
    await asyncio.gather(say('asdf'), say('hjkl'))
    await say(text='foo')
    await say(text='bar')


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

beginning at 3461039.617
say asdf at 3461040.618
say hjkl at 3461040.618
say foo at 3461041.620
say bar at 3461042.621

Answer 2

Not_a_Golfer 在评论中回答了我的问题。

将内部wraps()函数从协程更改为生成器解决了问题：

def run_method_in_executor(func, *, loop=None):
    def wraps(*args):
        _loop = loop if loop is not None else asyncio.get_event_loop()
        yield _loop.run_in_executor(executor=None, func=func, *args)
    return wraps

编辑： 这对 IO 非常有用，但我还没有弄清楚如何await生成执行器函数，这意味着如果我依赖装饰函数来更新我的任何一个使用的某些值，它将创建一个竞争条件其他异步函数。