前言

近期有收到朋友咨询关于如何给Fastapi框架加上相关类似的权限校验的装饰器处理器。然后我说推荐你直接使用注入依赖来实现比较好，但是他说可能装饰器看起来更好看，更独立一点。这个我竟无法反驳。由于之前做FLASK的时候确实都是基于装饰器的方式来校验权限，所以该来的还是会来的，下面我们就实现一下。

2 实现过程的记录

步骤1 定义有参装饰器

首先我们的知道权限校验装饰器，肯定是需要携带参数的，所以我们的需要定义一个有参装饰器。

def wrapper_auth(roles: Union[str, Sequence[str]] = None,
                 permissions: Union[str, Sequence[str]] = None):
    def wrapper(func):
        @functools.wraps(func)
        def _wrapper(*args: Any, **kwargs: Any) -> Response:
            print("内部函数！", roles)
            return func(*args, **kwargs)
        return _wrapper

    return wrapper

步骤2 装饰到路由上

@app.get('/')
@wrapper_auth
def auttlogin():
    return ''

步骤3 验证测试-咯嘣脆的异常

不出意外上面这种定义的装饰器的方式有问题，首先是或报

提示我们路由上缺少一个查询的参数名字就是func,也就是我们的装饰器上面的func，这就蛋疼了！那么再来一个处理，那我们允许func为空不就OK了吗！！！是吧！！！这样就不说话必要参数了！！！

步骤4 设置func为非必选参数

def wrapper_auth(roles: Union[str, Sequence[str]] = None,
                 permissions: Union[str, Sequence[str]] = None):
    def wrapper(func:Optional[Callable]=None):
        @functools.wraps(func)
        def _wrapper(*args: Any, **kwargs: Any) -> Response:
            print("内部函数！", roles)
            return func(*args, **kwargs)
        return _wrapper

    return wrapper

步骤5 在验证：再-咯嘣脆的异常

分析路由内部源码：

它需要是转化为一个是一个可调用的对象，所以我们的方面装饰写的方式就不对！！！原来的方式:

@app.get('/')
@wrapper_auth
def auttlogin():
    return ''

修改为：

@app.get('/')
@wrapper_auth()
def auttlogin():
    return ''

这样在测试就OK了！

但是还是有问题！！加入我装饰的是一个协程函数的呢？如下：

@app.get('/')
@wrapper_auth()
async def auttlogin():
    return ''

此时在测试:

好吧问题多多！！！那我就在继续判断处理，需要根据被装饰是什么来做不同的处理呗！！

步骤6 优化，同时支持同步或异步装饰

def wrapper_auth(roles: Union[str, Sequence[str]] = None,
                 permissions: Union[str, Sequence[str]] = None):


    def wrapper(func:Optional[Callable]=None):
        if asyncio.iscoroutinefunction(func):
            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Response:
                print("装饰协程函数")
                return await func(*args, **kwargs)
            return async_wrapper
        else:
            @functools.wraps(func)
            def _wrapper(*args: Any, **kwargs: Any) -> Response:
                print("装饰同步函数！", roles)
                return func(*args, **kwargs)
            return _wrapper

    return wrapper

然后在测试：ok!!!!

再进化！！我想当做装饰器也可以当做一个依赖项！

你妹d！奇葩需求真多！！！

既然如果是作为依赖项的话，那说明我我们的func肯定是没值的，那么我们的可以在装饰器制作做判断处理不就可以了！！！是不是！！！！！这不简单！！！

代码示例如下：

def wrapper_auth(roles: Union[str, Sequence[str]] = None,
                 permissions: Union[str, Sequence[str]] = None):


    def wrapper(func:Optional[Callable]=None):
        if func is None:
            print("被当做依赖项处理！！！")
            return True
        if asyncio.iscoroutinefunction(func):
            @functools.wraps(func)
            async def async_wrapper(*args: Any, **kwargs: Any) -> Response:
                print("装饰协程函数")
                return await func(*args, **kwargs)
            return async_wrapper
        else:
            @functools.wraps(func)
            def _wrapper(*args: Any, **kwargs: Any) -> Response:
                print("装饰同步函数！", roles)
                return func(*args, **kwargs)
            return _wrapper

    return wrapper


# @app.get('/')
# @wrapper_auth()
# async def auttlogin():
#     return ''


@app.get('/')
@wrapper_auth()
async def auttlogin(asuv=Depends(wrapper_auth)):
    return ''

然后测试验证：结果进入到装饰里面！！想想肯定是没办法进去的啊！！！最后改版！！

主要是需要注意，当做依赖项的时候，需要多重的（）（）的调用，最后一个性质和上面当做装饰的时候一样，而第一个则是为了进入到装饰器的内部，如果没有这个他不会进入到装饰器函数内部！

至此要做相关权限校验上面的自己在装饰器内部去发挥吧！！

以上仅仅是个人结合自己的实际需求，做学习的实践笔记！如有笔误！欢迎批评指正！感谢各位大佬！

结尾

END

简书：https://www.jianshu.com/u/d6960089b087

掘金：https://juejin.cn/user/2963939079225608

公众号：微信搜【程序员小钟同学】

小钟同学 | 文 【欢迎一起学习交流】| QQ：308711822