Python异步函数装饰器问题：main()为何不可调用及解决办法

问题背景 #

现有一个用于计算同步函数运行时间的装饰器：

import time
def decor(f):
  starttime=time.time()
  f()
  print("child function run time is ", (time.time()-starttime)*1000, "ms")
  return f

尝试将其用于装饰异步函数时：

import asyncio

async def sleep_and_print(seconds):
    print(f"starting async {seconds} sleep 😴")
    await asyncio.sleep(seconds)
    print(f"finished async {seconds} sleep ⏰")
    return seconds

@decor
async def main():
    results = await asyncio.gather(sleep_and_print(3), sleep_and_print(6))
    print(results)

asyncio.run(main())

会抛出 RuntimeWarning: coroutine 'main' was never awaited 错误。

若将装饰器改为异步函数并使用await：

async def decor(f):
  starttime=time.time()
  await f()
  print("child function run time is ", (time.time()-starttime)*1000, "ms")
  return f

调用asyncio.run(main())时又会出现 coroutine object is not callable 错误。

疑问 #

1. 为什么main()会变为不可调用对象？
2. 有哪些可行的解决办法？

原因分析 #

• 第一个同步装饰器的问题：直接调用异步函数f()会生成协程对象，但未通过await或asyncio.run()等待其执行，因此触发未等待协程的警告。
• 异步装饰器导致main不可调用的原因：当用异步函数作为装饰器时，@decor的执行逻辑是将main传入decor，执行decor(main)后返回的是协程对象，而非原本可调用的异步函数。后续调用main()时，本质是尝试调用一个协程对象，自然会抛出“coroutine object is not callable”错误。

可行解决办法 #

方法1：同步装饰器返回异步包装函数 #

装饰器本身为普通函数，内部定义异步包装函数，在包装函数中处理计时和异步执行：

import time
import asyncio

def async_time_decorator(f):
    async def wrapper(*args, **kwargs):
        start_time = time.time()
        result = await f(*args, **kwargs)
        elapsed = (time.time() - start_time) * 1000
        print(f"函数运行时间: {elapsed:.2f} ms")
        return result
    return wrapper

async def sleep_and_print(seconds):
    print(f"starting async {seconds} sleep 😴")
    await asyncio.sleep(seconds)
    print(f"finished async {seconds} sleep ⏰")
    return seconds

@async_time_decorator
async def main():
    results = await asyncio.gather(sleep_and_print(3), sleep_and_print(6))
    print(results)

asyncio.run(main())

方法2：同步装饰器内部直接运行协程 #

如果不需要被装饰函数保持异步特性，可在装饰器内部用asyncio.run()执行协程：

import time
import asyncio

def async_time_decorator(f):
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = asyncio.run(f(*args, **kwargs))
        elapsed = (time.time() - start_time) * 1000
        print(f"函数运行时间: {elapsed:.2f} ms")
        return result
    return wrapper

async def sleep_and_print(seconds):
    print(f"starting async {seconds} sleep 😴")
    await asyncio.sleep(seconds)
    print(f"finished async {seconds} sleep ⏰")
    return seconds

@async_time_decorator
async def main():
    results = await asyncio.gather(sleep_and_print(3), sleep_and_print(6))
    print(results)

main()

方法3：类实现异步装饰器 #

通过类的__call__方法实现装饰器，灵活性更高：

import time
import asyncio

class AsyncTimer:
    def __call__(self, func):
        async def wrapper(*args, **kwargs):
            start_time = time.time()
            result = await func(*args, **kwargs)
            elapsed = (time.time() - start_time) * 1000
            print(f"函数运行时间: {elapsed:.2f} ms")
            return result
        return wrapper

async def sleep_and_print(seconds):
    print(f"starting async {seconds} sleep 😴")
    await asyncio.sleep(seconds)
    print(f"finished async {seconds} sleep ⏰")
    return seconds

@AsyncTimer()
async def main():
    results = await asyncio.gather(sleep_and_print(3), sleep_and_print(6))
    print(results)

asyncio.run(main())

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