Python怎么测试异步接口

野性酷女 2023-10-09 10:10 75阅读 0赞

当业务处理比较耗时时, 接口一般会采用异步处理的方式, 这种异步处理的方式又叫Future模式.

**一般流程**  
当你请求一个异步接口,接口会立刻返回你一个结果告诉你已经开始处理,结果中一般会包含一个任务id类似的东西用于追踪结果, 另外会提供一个**查询结果**的接口, 当结果未处理完查询接口会返回相应的"未完成"状态, 如果已经处理完,则会返回相应的数据.

**处理方法**  
异步接口我们一般采取轮询的方法,每隔一定时间间隔取请求一下查询结果的接口,直到接口返回的状态是已完成/查询到指定数据或超时

如果异步接口没有提供追踪id和查询接口,我们可以通过同样的方法轮询查取数据库数据或日志数据直到获取到指定结果或超时

**示例接口**

1.  创建订单接口

**请求地址**  
`http://115.28.108.130:5000/api/order/create/`

**请求方法**  
GET/POST

**请求格式**  
POST: 表单

<table> 
 <thead> 
  <tr> 
   <th style="text-align:left;">参数</th> 
   <th>类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="text-align:left;">user_id</td> 
   <td>String</td> 
   <td>用户id</td> 
  </tr> 
  <tr> 
   <td style="text-align:left;">goods_id</td> 
   <td>String</td> 
   <td>商品id</td> 
  </tr> 
  <tr> 
   <td style="text-align:left;">num</td> 
   <td>int</td> 
   <td>数量</td> 
  </tr> 
  <tr> 
   <td style="text-align:left;">amount</td> 
   <td>float</td> 
   <td>总价</td> 
  </tr> 
 </tbody> 
</table>

**响应示例**  
缺少参数:

{
        "msg": "参数缺失"
    }

成功:

{
        "order_id": "69561"
    }

1.  获取订单结果接口

**请求地址**  
`http://115.28.108.130:5000/api/order/get_result/?order_id=***`

**请求方法**  
GET

<table> 
 <thead> 
  <tr> 
   <th style="text-align:left;">参数</th> 
   <th>类型</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="text-align:left;">order_id</td> 
   <td>String</td> 
   <td>订单id</td> 
  </tr> 
 </tbody> 
</table>

**响应示例**  
创建中:

{}

创建成功:

{
        "amount": "20.0",
        "goods_id": "123",
        "num": "2",
        "user_id": "123"
    }

**Python实现方法**

import time
    import requests
    
    def create_order():
        url = "http://115.28.108.130:5000/api/order/create/"   # 异步接口
        data = {
            "user_id": "1234",
            "goods_id": "136",
            "num": 10,
            "amount": 20.00
        }
        res = requests.post(url=url, data=data)
        return res.json().get("order_id")  # 返回order_id用于追踪
    
    def get_order_result(interval = 1, time_out = 60):  # 设置了默认时间间隔和超时时间,可以修改
        order_id = create_order()
        # 查询结果接口
        url = "http://115.28.108.130:5000/api/order/get_result/?order_id={}".format(order_id) 
        start_time = time.time()  # 启动时间
        end_time = start_time + time_out  启动时间+超时时间=结束时间
        count = 1  # 计数器, 此处是为了显示更直观, 可以去掉
        while time.time() < end_time:   # 当未到结束时间时, 循环请求
            res = requests.get(url)  # 请求查询结果接口
            print(count, res.json())
            count += 1
            time.sleep(interval)   # 休眠指定时间
            if res.json():   # 如果有数据则退出循环
                break
        else:
            return None   # 正常退出(达到end_time, 非break退出) 返回None
        return res.json()  # break退出返回 接口数据
    
    
    if __name__ == '__main__':
        order_result = get_order_result()
        print(order_result)

> 更多学习资料请加添加作者微信：lockingfree获取

转载于:https://www.cnblogs.com/superhin/p/11454965.html