所有异步、协程和同步功能的装饰器？答案

【问题标题】：Decorator for all async, coroutine and sync functions?所有异步、协程和同步功能的装饰器？
【发布时间】：2021-06-15 12:58:47
【问题描述】：

我为异步/协程和同步函数创建了以下装饰器。

def authorize(role):
    def decorator(f):

        @contextmanager
        def auth(): # Business logic shared by async and sync funtions
            if is_authorized(role): # role admin returns True, otherwise False
                yield
            else:
                print('ERROR')
                        
        @wraps(f)
        def wrapper(*args, **kwargs):
            if not asyncio.iscoroutinefunction(f):
                with auth():
                    return f(*args, **kwargs)
            else:
                async def tmp():
                    with auth():
                        return (await f(*args, **kwargs))
                return tmp()
        return wrapper
    return decorator

is_authorized() 返回 True 时效果很好。

@authorize(role='Readonly')
def test():
    print('TEST')

test()

但是，当is_authorized() 返回 False 时，它会引发异常。如果未授权，则不应调用修饰函数，它应返回 501 HTTP 错误。

@authorize(role='Readonly')
def test():
    print('TEST')

错误回溯（最近一次通话最后）：文件“”，第 1 行，在文件“”，第 13 行，在包装器中文件“C:\anaconda3\lib\contextlib.py”，第 115 行，在 __enter__ 中从无引发 RuntimeError("generator didn't yield") RuntimeError：生成器没有产生

【问题讨论】：

这里为什么需要contexmanager？当您需要释放资源，或者包装代码并抑制异常时，上下文管理器很有用，但不能替换 if 语句
下面的问题，contextmanager用于共享两种情况的逻辑，stackoverflow.com/questions/44169998/…
嗯，有，但也有不同的情况，你只需要 if 语句:-)

标签： python

【解决方案1】：

由于contextmanager 必须是生成器而引发的错误，这意味着它必须始终执行yield 语句，因为yield 将上下文管理器的__enter__ 和__exit__ 部分分开。在您的实现中，仅当is_autorized 返回True 时才会产生。

实际上你在这里不需要contexmanager，你需要简单的if 语句。

我通过参数传递is_authorized，因为它对于注入替代实现以进行测试或其他目的很有用。

import asyncio
import functools


def authorize(role, is_authorized):
    def decorator(f):
        @functools.wraps(f)
        def wrapper(*args, **kwargs):
            if is_authorized(role):
                if asyncio.iscoroutinefunction(f):
                    async def tmp():
                        return (await f(*args, **kwargs))
                    return tmp()
                else:
                    return f(*args, **kwargs)
            elif asyncio.iscoroutinefunction(f):
                # must return coro anyway
                async def tmp():
                    print("async unauthorized")
                    return None
                return tmp()
            else:
                print("sync unauthorized")
                return None
        return wrapper
    return decorator


def is_authorized(role):
    return role == "lucky"


@authorize("lucky", is_authorized)
async def func1():
    await asyncio.sleep(0)
    return "coro finished"

@authorize("whatever", is_authorized)
async def func2():
    await asyncio.sleep(0)
    return "coro would not called"

@authorize("lucky", is_authorized)
def func3():
    return "sync func finished"

@authorize("whatever", is_authorized)
def func4():
    return "would not called"


if __name__ == "__main__":
    print(asyncio.run(func1()))
    print(asyncio.run(func2()))
    print(func3())
    print(func4())

打印

coro finished

async unauthorized
None

sync func finished

sync unauthorized
None

【讨论】：