异步函数的类装饰器答案

【问题标题】：Class Decorator for Async Function异步函数的类装饰器
【发布时间】：2021-12-29 00:15:25
【问题描述】：

我希望标题不言自明，这里是代码：

class my_class:
    def foo(self,f):
        async def wrapper(*args, **kwargs):
            return await f(*args, **kwargs)
        return wrapper

inst = my_class()
@inst.foo
async def this():
    print("This is cool")

task = asyncio.create_task(this())

如您所见，我正在尝试使用装饰器 foo（来自 my_class）运行函数 this()。

这是错误：

Traceback (most recent call last):
  File "<censored>", line 23, in <module>
    task = asyncio.create_task(this())
  File "/Library/Frameworks/Python.framework/Versions/3.9/lib/python3.9/asyncio/tasks.py", line 360, in create_task
    loop = events.get_running_loop()
RuntimeError: no running event loop
sys:1: RuntimeWarning: coroutine 'my_class.foo.<locals>.wrapper' was never awaited

【问题讨论】：

标签： python-3.x python-asyncio python-decorators

【解决方案1】：

你基本上就在那里。只需将asyncio.create_task 替换为asyncio.run。

class my_class:
def foo(self,f):
    async def wrapper(*args, **kwargs):
        print("inside the wrapper")
        return await f(*args, **kwargs)
    return wrapper


inst = my_class()
@inst.foo
async def this():
    print("This is cool")


asyncio.run(this())
# inside the wrapper
# This is cool

# or...
async def main():
    this()


if __name__ == "__main__":
    asyncio.run(main())

asyncio.create_task，在asyncio.get_running_loop() 上创建一个任务。它不会等待它运行。它返回一个Task，可以是await-ed。为您的用例保持简单。 :)

【讨论】：