Python类实例方法作异步函数装饰器报无运行事件循环如何解决？

你遇到的两个报错/警告的根因分别是：

• RuntimeError: no running event loop：asyncio.create_task() 要求调用时必须存在正在运行的事件循环，你直接在全局同步上下文里调用该方法，此时还没有启动任何事件循环，因此触发报错。
• 协程未等待的警告：因为上一步创建task失败，装饰器返回的wrapper协程对象没有被调度执行、也没有被手动await，解释器退出时就会抛出这个警告。

注意：你当前写的类实例方法作为异步装饰器的核心逻辑是没有问题的，报错和装饰器本身的实现无关，是异步函数的调用方式错误导致的。

方案1：使用asyncio.run直接运行顶层协程（Python 3.7+ 推荐） #

asyncio.run() 会自动创建并启动事件循环，运行传入的顶层协程，结束后自动关闭循环，是最简单的调用方式：

import asyncio

class my_class:
    def foo(self,f):
        async def wrapper(*args, **kwargs):
            # 这里可以加装饰器的自定义逻辑，比如前置检查、日志打印等
            print("装饰器前置逻辑")
            res = await f(*args, **kwargs)
            print("装饰器后置逻辑")
            return res
        return wrapper

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

# 直接用run启动协程
asyncio.run(this())

方案2：手动管理事件循环，提前创建task #

如果你确实需要先创建task再调度运行，可以手动获取事件循环，在循环上下文中处理：

import asyncio

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")

loop = asyncio.get_event_loop()
# 在循环内创建task并运行
task = loop.create_task(this())
loop.run_until_complete(task)
loop.close()

运行上述两种方案的代码都可以正常输出预期内容，类实例方法作为异步装饰器的功能可以正常生效。

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