使用协程作为装饰器

Question

在这种情况下：

async def foo(f):
    async def wrapper(*args, **kwargs):
        return f(*args, **kwargs)
    return wrapper

@foo
async def boo(*args, **kwargs):
    pass

对 foo 作为 boo 装饰器的装饰器的调用是异步调用吗？

--第一次编辑： 另外，如何处理作为装饰器的协程调用链？

Answer 1

感谢@blacknght的评论，考虑到

def foo():
    def wrapper(func):
        @functools.wraps(func)
        async def wrapped(*args):
             # Some fancy foo stuff
            return await func(*args)
        return wrapped
    return wrapper

和

def boo():
    def wrapper(func):
        @functools.wraps(func)
        async def wrapped(*args):
            # Some fancy boo stuff
            return await func(*args)
        return wrapped
    return wrapper

作为两个装饰器，并且

@foo()
@boo()
async def work(*args):
    pass

由于

foo

正在包装

work

协程，因此关键在于

await

两个装饰器中的

func(*arg)

。

Answer 2

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

@foo
async def boo(*args, **kwargs):
    pass

你的装饰器需要是一个普通的函数，它就能正常工作。

当评估装饰器时，Python 会以函数作为参数执行该方法。

@foo
async def boo():
    pass

评估为：

__main__.boo = foo(boo)

如果 foo 是一个异步函数类型（main.boo）将是一个协程对象，而不是一个函数对象。但如果 foo 是常规同步函数，它将立即求值，并且 main.boo 将是返回的包装器。

Answer 3

这是使用

decorator

库的替代方法（即首先使用

pip install decorator

）：

import asyncio

import decorator


@decorator.decorator
async def decorate_coro(coro, *args, **kwargs):
    try:
        res = await coro(*args, **kwargs)
    except Exception as e:
        print(e)
    else:
        print(res)


@decorate_coro
async def f():
    return 42


@decorate_coro
async def g():
    return 1 / 0


async def main():
    return await asyncio.gather(f(), g())

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

输出：

42
division by zero

Answer 4

适用于同步和异步函数的装饰器：

from inspect import iscoroutinefunction
from functools import wraps

def my_decorator(decorator_option=None):
    def decorator(f):
        def shared_logic(*args, **kwargs):
            # Do something with decorator_option, if relevant
            pass

        @wraps(f)
        def wrapper(*args, **kwargs):
            shared_logic(*args, **kwargs)
            return f(*args, **kwargs)

        @wraps(f)
        async def async_wrapper(*args, **kwargs):
            shared_logic(*args, **kwargs)
            return await f(*args, **kwargs)

        return async_wrapper if iscoroutinefunction(f) else wrapper
    return decorator

如果装饰器不带任何参数，则可以删除外部函数。

Answer 5

async def foo(f):
  async def wrapper(*args, **kwargs):
    # wrapper pre-function stuff
    result = await f(*args, **kwargs) # key is to await function's result
    # wrapper post-function stuff
    return result 
  wrapper.__name__ = f.__name__ # for some reason, async wrappers don't do this
  # do it to avoid an error if you use the wrapper on multiple functions
  return wrapper

两个关键的更改是等待您正在包装的函数（因为它是一个异步函数），以及更改包装器函数的名称，以便您的程序不会尝试将多个函数命名为相同的东西。