httprunner3源码解读（3）client.py httprunner3源码解读（3）client.

源码目录结构
文章图片

ApiResponse 这个类没啥好说的

class ApiResponse(Response): """ 继承了requests模块中的Response类，重写了里面的raise_for_status方法 """ def raise_for_status(self): if hasattr(self, "error") and self.error: raise self.error Response.raise_for_status(self)

get_req_resp_record 这个函数的功能是获取请求记录和响应记录，源码分为4段来看

第1段

def get_req_resp_record(resp_obj: Response) -> ReqRespData: """ :param resp_obj: Response响应 :return: 返回自定义的ReqResData模型类 """def log_print(req_or_resp, r_type): """ 日志打印，格式为标准的json """ msg = f"\n================== {r_type} details ==================\n" for key, value in req_or_resp.dict().items(): # 如果value中还包含着dict或者list，就把value转成json格式 if isinstance(value, dict) or isinstance(value, list): value = https://www.it610.com/article/json.dumps(value, indent=4, ensure_ascii=False)msg +="{:<8} : {}\n".format(key, value) logger.debug(msg)

第1段代码就是定义了一个打印日志的函数，打印的日志解析为标准的json格式

第2段

# 记录实际请求信息（请求头、cookie信息、请求体） request_headers = dict(resp_obj.request.headers) request_cookies = resp_obj.request._cookies.get_dict()request_body = resp_obj.request.body if request_body is not None: try: request_body = json.loads(request_body) except json.JSONDecodeError: # str: a=1&b=2 pass except UnicodeDecodeError: # bytes/bytearray: request body in protobuf pass except TypeError: # neither str nor bytes/bytearray, e.g. pass# lower_dict_keys的作用是将字典中的key大写转小写 request_content_type = lower_dict_keys(request_headers).get("content-type") if request_content_type and "multipart/form-data" in request_content_type: # upload file type request_body = "upload file stream (OMITTED)"request_data = https://www.it610.com/article/RequestData( method=resp_obj.request.method, url=resp_obj.request.url, headers=request_headers, cookies=request_cookies, body=request_body, ) # 在debug模式下打印请求日志 log_print(request_data,"request")

第2段代码是先获取request_headers、request_cookies、request_body，然后将获取到的信息放入RequestData模型中，最后打印请求的信息

第3段

# 记录响应信息 resp_headers = dict(resp_obj.headers) lower_resp_headers = lower_dict_keys(resp_headers) content_type = lower_resp_headers.get("content-type", "")if "image" in content_type: # response is image type, record bytes content only response_body = resp_obj.content else: try: # try to record json data response_body = resp_obj.json() except ValueError: # only record at most 512 text charactors resp_text = resp_obj.text response_body = omit_long_data(resp_text)response_data = https://www.it610.com/article/ResponseData( status_code=resp_obj.status_code, cookies=resp_obj.cookies or {}, encoding=resp_obj.encoding, headers=resp_headers, content_type=content_type, body=response_body, )# 在debug模式下打印响应日志 log_print(response_data,"response")

第3段代码是获取resp_headers、content_type、response_body，最后将这些数据都放入ResponseData模型类中，最后打印响应日志

第4段

req_resp_data = https://www.it610.com/article/ReqRespData(request=request_data, response=response_data) return req_resp_data

最后这段就是将刚才的请求信息和响应信息全部放入ReqRespData模型中，最后get_req_resp_record函数返回的内容就是ReqRespData模型

HttpSession 在requests.Session上进行了二次封装，该类包含4个方法，下面依次介绍

init

def __init__(self): super(HttpSession, self).__init__() self.data = https://www.it610.com/article/SessionData()

初始化方法，定义了data属性的默认值为SessionData模型，该模型包含了req_resps: List[ReqRespData] = []请求响应内容

update_last_req_resp_record

def update_last_req_resp_record(self, resp_obj): """ update request and response info from Response() object. """ # TODO: fix self.data.req_resps.pop() self.data.req_resps.append(get_req_resp_record(resp_obj))

更新最新的请求响应记录，放入req_resps列表中

request 发送requests.Request请求，返回requests.Response响应，还做了以下事情

1.设置了超时时间120s
2.计算整个请求花费了多少时间
3.定义了客户端ip地址和端口号、服务端ip地址和端口号
4.计算了响应体的内容大小
5.记录了消耗时间
6.记录了request和response记录，包括重定向记录

_send_request_safe_mode 【httprunner3源码解读（3）client.py】发送一个http请求，并捕获由于连接问题可能发生的任何异常

def _send_request_safe_mode(self, method, url, **kwargs): """ Send a HTTP request, and catch any exception that might occur due to connection problems. Safe mode has been removed from requests 1.x. """ try: return requests.Session.request(self, method, url, **kwargs) except (MissingSchema, InvalidSchema, InvalidURL): raise except RequestException as ex: resp = ApiResponse() resp.error = ex resp.status_code = 0# with this status_code, content returns None resp.request = Request(method, url).prepare() return resp