Python + Requests 实现接口自动化流程

# create by: wyun
# create at: 2020/4/19 9:30
import json

import requests


class RequestBase:

    def __init__(self, request_type, request_url, request_body, request_headers, intf_type=None):
        """
        请求接口公共类
        :param request_type: 请求类型：post, get
        :param request_url: 请求url地址
        :param request_headers:  请求头
        :param request_body:  请求体（xml类型传入字符串格式，json类型数据必须传入json格式，不能传入字典）
        :param intf_type:  接口类型：webservice， api，分别对应接口数据格式：xml， json
        """
        self.request_type = request_type
        self.request_url = request_url
        self.request_body = request_body
        self.request_headers = request_headers

        self.intf_type = intf_type
        self.res = self.__send_request()

    # 发送请求
    def __send_request(self):
        if not isinstance(self.request_type, str):
            print(‘请求类型格式错误。‘)
            return None
        if self.request_type.upper() not in [‘GET‘, ‘POST‘, ‘HEAD‘, ‘OPTIONS‘, ‘PUT‘, ‘DELETE‘, ‘TRACE‘, ‘CONNECT‘]:
            print(‘请求类型不存在。‘)
            return None
        return requests.request(method=self.request_type, url=self.request_url, data=self.request_body.encode(‘utf-8‘),
                               headers=self.request_headers)

    # 获取响应
    def get_respond(self):
        return self.res

    # 获取响应头
    def get_respond_head(self):
        return self.res.headers

    # 获取响应体
    def get_respond_body(self):
        if self.res.content:
            return self.res.text

# create by: wyun
# create at: 2020/4/19 12:58
import json
import re

"""
对请求体或响应体进行处理：
1. 支持读取请求体或响应体中字段的值
2. 支持更新请求体中字段的值
"""


# 递归调用，更新json中的字段值
def update_json_step(node, json_str, value, i=0):
    # 当前节点索引（负向）
    node_index = -len(node) + i

    # 如果包含[n]形式，说明节点为列表，需处理
    if ‘[‘ in node[node_index] and ‘]‘ in node[node_index]:
        list_node, list_index = node[node_index].split(‘[‘)
        index = list_index.split(‘]‘)[0]
        if index is None or index == ‘‘:
            print(‘参数传入错误，请指定列表[%s]索引‘ % list_node)
        else:
            index = int(index)
        return update_json_step(node, json_str[list_node][index], value, i + 1)

    # 判断如果当前节点为最后一个节点，则更新value值
    if node_index == -1:
        json_str[node[-1]] = value
        return json_str

    return update_json_step(node, json_str[node[node_index]], value, i + 1)

class DataHandle:

    def __init__(self, data_type, data_msg, fields, value_dict=None):
        """
        处理请求体或响应体数据，读取或更新字段值
        :param data_type:  数据类型：xml，json
        :param data_msg: 请求体或响应体
        :param fields: 字段值，支持多字读方式，字段间逗号隔开；或传入列表、元组
                        json格式：需写入完整节点路径，如：body.base.name，对应list类型的需传入索引位置，如：body.baselist[0].name
        :param value_dict: 以字典键值对保存字段值
        """
        self.data_type = data_type
        self.data = data_msg
        # fields支持str，list方式，str自动转换为list
        if isinstance(fields, str):
            self.fields = fields
            self.fields_list = []
            if ‘,‘ in fields:
                self.fields_list = fields.strip().split(‘,‘)
            else:
                self.fields_list.append(fields.strip())
        elif isinstance(fields, list):
            self.fields_list = fields
        else:
            self.fields_list = list(fields)
        # 初始化字典值
        if value_dict is None:
            self.value_dict = dict()
        else:
            self.value_dict = value_dict

    def get_fields_value(self):
        """
        获取字段值
        1. 支持获取多个字段值，输入字符串或列表， 元组
        """
        step_dict = self.value_dict
        # 字段列表循环获取
        for p in self.fields_list:
            if p == ‘‘:
                continue
            # 处理xml格式报文
            if self.data_type.upper() == ‘XML‘:
                pattern = ‘<‘ + p + ‘>.*</‘ + p + ‘>‘
                search_result = re.search(pattern, self.data)
                if search_result is not None:
                    # 包括标签和值都匹配上
                    field_and_value = search_result.group()
                    # 去除标签获取字段值，并存入字典
                    step_dict[p] = (field_and_value.split(‘</‘)[0]).split(‘>‘)[-1]
                else:
                    # print(‘参数[%s]提取失败，无匹配值。‘ % p)
                    pass
            # 处理json格式报文
            elif self.data_type.upper() == ‘JSON‘:
                # 获取层级(使用.隔开)
                node = p.split(‘.‘)
                # json转换字典
                if isinstance(self.data, dict):
                    temp = self.data
                else:
                    temp = json.loads(self.data)

                # 逐层读取数据
                for per_node in node:
                    if ‘[‘ in per_node and ‘]‘ in per_node:
                        per_node, index_str = per_node.split(‘[‘)
                        index = int(index_str.split(‘]‘)[0])                        
temp = temp.get(per_node)[index]                        
if temp is None:                            
break                    
else:                        
temp = temp.get(per_node)                        
if temp is None:                            
break                
# 讲读取的结果写入字典                
step_dict[p] = temp            
# 非 xml，json的数据格式，报错退出            
else:                
print(‘ERROR: 不支持此数据类型[%s]‘ % self.data_type)                
break        
return step_dict    

def update_fields_value(self):        
"""        更新字段值        支持更新单个字段值，多个字段值（字段名 字符串或列表、元组，字段值 字典）        """        



req_body = self.data        
# 字段列表循环更新        
for p in self.fields_list:            
# 入参类型判断            
if isinstance(self.value_dict, dict):                
# 参数在字典中不存在，则跳过                
if p not in self.value_dict.keys():                    
print(‘ERROR： 字典中不存在参数[%s]。‘ % p)                    
continue                
# 获取字典中参数的值                
value = self.value_dict[p]            
elif isinstance(self.value_dict, str):                
value = self.value_dict            
else:                
print(‘函数入参[%s]类型不支持，请传入字符串或字典。‘ % self.value_dict)                
break            
# 更新数据类型为 xml 的参数值            
if self.data_type.upper() == ‘XML‘:                
# 优先寻找是否存在指定更新参数，若有，则按指定参数更新                
if ‘${‘ + p + ‘}‘ in req_body:                    
req_body = req_body.replace(‘${‘ + p + ‘}‘, value)                
# 其次寻找标签参数值，若有，则按标签更新值                
else:                    
# 检查标签是否存在，若存在                    
if ‘<‘ + p + ‘>‘ in req_body and ‘</‘ + p + ‘>‘ in req_body:                        
# 正则匹配                        
pattern = ‘<‘ + p + ‘>.*</‘ + p + ‘>‘                        
new_field_and_value = ‘<‘ + p + ‘>‘ + str(value) + ‘</‘ + p + ‘>‘                        
search_result = re.search(pattern, req_body)                        
# 若正则匹配结果存在，则进行字段值更新                        
if search_result is not None:                            
old_field_and_value = search_result.group()                            
req_body = req_body.replace(old_field_and_value, new_field_and_value)                        
else:                            
print(‘参数[%s]匹配标签失败，请检查报文中标签格式。‘ % p)                    
else:                        
# print(‘参数[%s]匹配标签失败，无此标签。‘ % p)                        
continue            
# 更新数据类型为 xml 的参数值            
elif self.data_type.upper() == ‘JSON‘:                
# json转换字典                
if isinstance(self.data, dict):                    
req_body = self.data                
else:                    
req_body = json.loads(self.data)                
# 获取层级                
node = p.split(‘.‘)                
# 调用递归函数更新字段值                
update_json_step(node, req_body, value)            

# 非 xml，json的数据格式，报错退出            
else:                
print(‘ERROR: 不支持此数据类型[%s]‘ % self.data_type)                
break        

return req_body

# create by: wyun
# create at: 2020/4/23 22:42
from api.comm.data_handle import DataHandle


def intf_base_init(data_type, request_body, **kwargs):
    """
    接口初始化通用类
    :param data_type: 数据类型，如：xml，json
    :param request_body: 请求体
    :param kwargs: 可变关键字参数
    :return: 请求体
    """
    return DataHandle(data_type, request_body, kwargs.keys(), kwargs).update_fields_value()