测试平台系列(65) 异步方法装饰器

大家好~我是米洛！

我在从0到1打造一个开源平台, 也在编写一套完整的接口测试平台系列教程，希望大家能够多多支持。

欢迎关注我的公众号测试开发坑货，获取最新文章教程!

回顾

上一节我们完善了一整套测试用例执行的流程，这一节我们来讲讲async方法的装饰器。

在此之前，我们先来看下用于异步方法的装饰器。

日志装饰器

还记得我们之前编写过的日志装饰器吗？不记得的话也没关系，咱们现卤一个。

import functools
def log(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        print("正在执行方法: ", func.__name__)
        print("参数: ", *args, *[f"{k}={v} " for k, v in kwargs.items()])
        result = func(*args, **kwargs)
        print("执行结果: ", result)
        return result
    return wrapper
@log
def print_user(name, age, height=180):
    return f"hello, {name}. Age: {age}, Height: {height}"
if __name__ == "__main__":
    print_user("klose", 43, height=182)

我们编写了一个print_user(打印用户信息)的方法，并给他加上了log装饰器，这样一旦这个方法执行了，我们就会输出这些数据到日志里面。

因为这里我没有现成的日志包，所以我用了print代替。输入到日志需要记录什么内容呢？

• 调用了什么方法

• 方法的参数是什么

• 方法的返回值是什么
  我们来看看输出:

就是这么个场景，普通的装饰器实现如上。接着我们来思考一个问题，如果我们的方法是异步的，在装饰器不变的情况下，会是什么输出呢？

求豆麻袋。

我们来进行下简单的改造。

async方法的装饰器

@log
async def print_user(name, age, height=180):
    return f"hello, {name}. Age: {age}, Height: {height}"

只需要把print_user方法改为async，然后用asyncio.run执行print_user方法即可。来看下输出:

@log
async def print_user(name, age, height=180):
    return f"hello, {name}. Age: {age}, Height: {height}"
if __name__ == "__main__":
    asyncio.run(print_user("klose", 43, height=182))

可以看到，执行结果变成了coroutine了，也就是说拿不到返回值了。

那么接下来就进入我们的改造阶段，让装饰器也支持异步方法。

查询资料

通过google，我们发现asyncio有这样一个方法: iscoroutinefunction

看名字就知道，是判断一个function是不是coroutine，如果这招有用的话，那说明我们的问题能够解决了。

def log(func):
    if asyncio.iscoroutinefunction(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            print("正在执行方法: ", func.__name__)
            print("参数: ", *args, *[f"{k}={v} " for k, v in kwargs.items()])
            result = await func(*args, **kwargs)
            print("执行结果: ", result)
            return result
    else:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            print("正在执行方法: ", func.__name__)
            print("参数: ", *args, *[f"{k}={v} " for k, v in kwargs.items()])
            result = func(*args, **kwargs)
            print("执行结果: ", result)
            return result
    return wrapper

我们把代码改成这个样子: 可以看到，装饰器又和上一次讲的一样，进入了一个分水岭，当iscoroutine成立的时候，形成了一个async的装饰器，反之则是普通的装饰器。

试试效果先

再试试把log放到普通方法上面:

代码冗余

细心的朋友可能发现了，3个print的内容都是很重复的，但是ide并没有提示你，但其实这几块内容是完全可以封装起来的。

这大概就是顺手为之，慢慢就形成了屎山代码吧~

所以核心方法是asyncio.iscoroutinefunction, 你学会了吗？

今天的内容就介绍到这里了，大家没点关注的点个关注，点了关注的点个赞，点了赞的点个在看，点了在看的点个赞赏，点了赞赏的点个退出。