requests封装：让HTTP服务人类

大家好~我是小方，欢迎大家关注「测试开发笋货」体完记得俾个「like」呀

失踪人口回归

自从上一期更新后，差不多一个月都没更新了，最近一个月差不多是躺平的状态吧，挺舒服的~马上年底，年底之后就过年了，不能再继续躺平了，懂的都懂

requests封装

大家在做接口自动化或者写造数脚本，与api交互的大多使用的是requests，既然用得这么频繁，咱们进行一下封装处理吧~代码借鉴于无敌哥的pity~

关键代码解析

首先实例化对象，是否使用会话请求，如果使用会话请求就创建会话对象，会话请求时，统一请求头

接着发送请求，固定传参method、url和判断标识，kwarges为不定传参

返参解析，传入返参对象，判断标识，因为统一了接口返参数据结构，会有部分老旧接口不适配，这时候ignore字段传入false即不校验数据结构

其余方法可看完整代码

import requests
import json
from common.bm_log import mylog

class BmRequest(object):

    def __init__(self, session=False, headers=None):
        """
        初始化
        :param session: 是否使用会话请求
        :param headers: 统一请求头
        """
        self.session = session
        if self.session :
            # 创建会话对象
            self.client = requests.Session()
            if headers: self.client.headers.update(headers)
            return
        self.client = requests
        self.response = None


    def send(self, method, url, ignore=True, **kwargs):
        """
        发送请求的方法
        :param ignore: 返参是否进行强判断
        :param method: 请求方法
        :param url: 请求url
        :param kwargs: headers请求头字典, data、json、files
        :return:
        """
        method = method.upper()

        kwargs["json"] = self.params("json", kwargs)
        kwargs["data"] = self.params("data", kwargs)
        self.response = self.client.request(method, url, timeout=10, **kwargs)
        status_code = self.response.status_code
        if status_code != 200:
            mylog.error(f"http请求报错, HTTP Status:{status_code}, url: {url}, method: {method}, params: {kwargs}, response: {self.response.text}")
            raise Exception(f"http请求报错, HTTP Status:{status_code}")
        response_data = self.get_response(self.response, url, kwargs, ignore)
        return response_data


    def get_response(self, response, url, kwargs, ignore=True):
        """
        :param response: 返参对象
        :param ignore: 是否进行强判断, 默认进行强判断
        :return:
        """
        try:
            data = response.json()
        except:
            return response.text
        if not ignore: return data
        if int(data.get('responseCode'))== 0 and data.get('isSuccess') == True:
            return data
        else:
            raise Exception(f"接口异常, url: {url}, parmas: {json.dumps(kwargs, ensure_ascii=False)}, response: {json.dumps(data, ensure_ascii=False)}")

    @staticmethod
    def params(param_name, param_dict):
        if param_name in param_dict:
            data = param_dict.get(param_name)
            if isinstance(data, str):
                try:
                    data = json.loads(data)
                except:
                    data = eval(data)
            return data

    def request_headers(self):
        # 获取请求头
        return self.response.request.headers

    def response_headers(self):
        # 获取返参头
        return self.response.headers

    def cookies(self):
        # 获取cookies
        return dict(self.response.cookies)
    def update_headers(self, headers):
        #更新会话的请求头
        self.client.headers.update(headers)

来测试一下呗

这里的api来自于无敌哥的pity 项目地址：http://test.pity.fun/

不使用会话请求

if __name__ == '__main__':
    auth_url = 'http://test.pity.fun/auth'
    login_json = {"username":"woody","password":"wuranxu"}
    sessioon = BmRequest()
    res_data = sessioon.send('post', auth_url+'/login', json=login_json)
    print(res_data)
    project_data = sessioon.send('get',auth_url+'/listUser', headers = dict(token=res_data['data']['token']))
    print(project_data)

出现报错了：
if int(data.get('responseCode'))== 0 and data.get('isSuccess') == True:
TypeError: int() argument must be a string, a bytes-like object or a number, not 'NoneType'
上面说了，要适配公司的数据结构，相应地改一下代码

使用会话请求

if __name__ == '__main__':
    auth_url = 'http://test.pity.fun/auth'
    login_json = {"username": "woody", "password": "wuranxu"}
    header = {"Content-Type": "application/json", "User-Agent": "wudige niubi"}
    sessioon = BmRequest(session=True, headers=header)
    res_data = sessioon.send('post', auth_url + '/login', json=login_json)
    print(res_data)
    sessioon.update_headers(dict(token=res_data['data']['token']))
    user_data = sessioon.send('get', auth_url + '/listUser')
    print(user_data)
    print(sessioon.request_headers())