测试平台系列(65) 异步方法装饰器-阿里云开发者社区

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

2022-05-30 241

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

日志服务 SLS，月写入数据量 50GB 1个月

简介： 异步方法装饰器

大家好~我是米洛！

我在从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))

image

可以看到，执行结果变成了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