bug fix: 修复HTTP请求对application/json数据格式的支持
This commit is contained in:
azhengzz
parent
f0ebdb6599
commit
39d217e91a
|
|
@ -18,7 +18,8 @@ import re
|
|||
import time
|
||||
|
||||
from app.cores.dictionaries import CONTENT_TYPE
|
||||
from app.cores.case.http.request_util import handle_params, handle_data, handle_file_upload, save_binary_to_file
|
||||
from app.cores.case.http.request_util import handle_params, handle_data_for_content_type, handle_file_upload, \
|
||||
save_binary_to_file, handle_data_form_json_str, handle_data_from_params
|
||||
|
||||
|
||||
class HTTPRequest:
|
||||
|
|
@ -142,7 +143,7 @@ class HTTPRequest:
|
|||
'status_code': self.response_status_code,
|
||||
'method': self.response.request.method,
|
||||
'url': self.response.url,
|
||||
'data': str(self.response.request.body), # 当时文件上传请求时response.request.body类型为bytes,需要转化为str
|
||||
'data': str(self.response.request.body), # 文件上传请求response.request.body类型为bytes,需要转化为str
|
||||
}
|
||||
|
||||
def _handle_request_headers(self):
|
||||
|
|
@ -208,37 +209,52 @@ def make_request(method, url, *, parameters=None, content_type=None, message_bod
|
|||
:type kwargs: dict
|
||||
:param headers: 请求头数据
|
||||
:type headers: dict
|
||||
|
||||
:rtype: HTTPRequest
|
||||
"""
|
||||
# 选择适当的参数数据, 如果parameters没有数据，则取message_body作为入参数据
|
||||
params = []
|
||||
|
||||
kwargs.pop('url', None)
|
||||
headers = kwargs.pop('headers', {})
|
||||
files = kwargs.pop('files', None)
|
||||
_json = kwargs.pop('json', None)
|
||||
_data = kwargs.pop('data', None)
|
||||
kwargs.pop('params', None)
|
||||
|
||||
# 选择适当的参数数据, 如果parameters没有数据，则取message_body作为入参数据。（参数和消息体有且只有一个生效）
|
||||
if (parameters is not None) and (len(parameters) > 0):
|
||||
params = parameters
|
||||
_data = handle_data_from_params(parameters)
|
||||
elif (message_body is not None) and (message_body.strip() != ''):
|
||||
params = message_body
|
||||
try:
|
||||
_json = handle_data_form_json_str(json_str=message_body)
|
||||
except json.decoder.JSONDecodeError:
|
||||
_data = message_body
|
||||
|
||||
# 如果类型错误是否编译通过? 使用rtype即使类型错误也可以编译通过。但如果使用type hints如果类型标识错误则编译不通过
|
||||
if method.lower() == 'get':
|
||||
return Get(url=url, params=handle_params(params), **kwargs)
|
||||
params = handle_params(parameters)
|
||||
return Get(url=url, params=params, data=None, json=None, headers=headers, files=None, **kwargs)
|
||||
if method.lower() == 'post':
|
||||
headers = kwargs.pop('headers', {})
|
||||
files = handle_file_upload(file_upload) if file_upload else None
|
||||
# TODO 如果勾选了对"post使用multipart/form-data"而参数数据为空, 消息体数据有json字符串, 则请求头的content-type会置为application/json
|
||||
# files = handle_file_upload(file_upload) if file_upload else None
|
||||
if content_type == CONTENT_TYPE.FORM_DATA:
|
||||
data = MultipartEncoder(handle_data(params, content_type=CONTENT_TYPE.FORM_DATA))
|
||||
headers['Content-Type'] = data.content_type
|
||||
else:
|
||||
data = handle_data(params)
|
||||
return Post(url=url, data=data, files=files, headers=headers, **kwargs)
|
||||
_json = None
|
||||
if parameters:
|
||||
_data = handle_data_for_content_type(parameters, content_type=CONTENT_TYPE.FORM_DATA)
|
||||
# TODO 支持文件上传可使用下列方法
|
||||
# _data.update({'file': ('filename', open(r'filepath', 'rb'), 'text/plain')})
|
||||
_data = MultipartEncoder(_data)
|
||||
if 'Content-Type' not in headers:
|
||||
headers['Content-Type'] = _data.content_type
|
||||
return Post(url=url, params=None, data=_data, json=_json, headers=headers, files=files, **kwargs)
|
||||
if method.lower() == 'delete':
|
||||
return Delete(url=url, data=handle_data(params), **kwargs)
|
||||
return Delete(url=url, params=None, data=_data, json=_json, headers=headers, files=None, **kwargs)
|
||||
if method.lower() == 'put':
|
||||
return Put(url=url, data=handle_data(params), **kwargs)
|
||||
return Put(url=url, params=None, data=_data, json=_json, headers=headers, files=None, **kwargs)
|
||||
if method.lower() == 'patch':
|
||||
return Patch(url=url, data=handle_data(params), **kwargs)
|
||||
return Patch(url=url, params=None, data=_data, json=_json, headers=headers, files=None, **kwargs)
|
||||
if method.lower() == 'options':
|
||||
return Options(url=url, **kwargs)
|
||||
return Options(url=url, params=None, data=_data, json=_json, headers=headers, files=None, **kwargs)
|
||||
if method.lower() == 'head':
|
||||
return Head(url=url, **kwargs)
|
||||
return Head(url=url, params=None, data=_data, json=_json, headers=headers, files=None, **kwargs)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
|
|
|||
|
|
@ -10,7 +10,7 @@ LOG:
|
|||
"""
|
||||
|
||||
from flask_login import current_user
|
||||
from typing import Iterable
|
||||
from typing import Iterable, Tuple, Dict
|
||||
import json
|
||||
import os
|
||||
|
||||
|
|
@ -20,9 +20,7 @@ from app.cores.dictionaries import CONTENT_TYPE
|
|||
def handle_params(params):
|
||||
"""处理请求参数，使其满足get请求查询字符串"""
|
||||
ret = {}
|
||||
if isinstance(params, str):
|
||||
ret = json.loads(params)
|
||||
elif isinstance(params, Iterable):
|
||||
if isinstance(params, Iterable):
|
||||
for param in params:
|
||||
if param.name_ in ret.keys():
|
||||
ret[param.name_].append(param.value_)
|
||||
|
|
@ -43,21 +41,42 @@ def handle_headers(headers):
|
|||
return ret
|
||||
|
||||
|
||||
def handle_data(params, content_type=None):
|
||||
def handle_data_from_params(params):
|
||||
"""
|
||||
处理请求参数，使其满足post请求的数据格式
|
||||
:param params: 参数数据
|
||||
:type params: Iterable[HTTPCaseParameter]
|
||||
:return: 处理后的参数数据
|
||||
:rtype: Dict
|
||||
"""
|
||||
if isinstance(params, Iterable):
|
||||
# return tuple((param.name_, param.value_) for param in params)
|
||||
return {param.name_: param.value_ for param in params}
|
||||
|
||||
|
||||
def handle_data_for_content_type(params, content_type=None):
|
||||
"""
|
||||
处理请求参数，使其满足post请求的数据格式
|
||||
:param params: 参数数据
|
||||
:type params: Iterable[HTTPCaseParameter]
|
||||
:param content_type: HTTP请求体数据类型
|
||||
:type content_type: str
|
||||
:return: 处理后的参数数据
|
||||
:rtype: Dict
|
||||
"""
|
||||
# return [(param.name, param.value) for param in params]
|
||||
if isinstance(params, str):
|
||||
return tuple(json.loads(params).items())
|
||||
elif isinstance(params, Iterable):
|
||||
if isinstance(params, Iterable):
|
||||
if content_type == CONTENT_TYPE.FORM_DATA:
|
||||
return tuple((param.name_, (param.name_, param.value_, param.content_type)) for param in params)
|
||||
return tuple((param.name_, param.value_) for param in params)
|
||||
return {param.name_: param.value_ for param in params}
|
||||
|
||||
|
||||
def handle_data_form_json_str(json_str):
|
||||
"""
|
||||
处理消息体json字符串，转为py对象
|
||||
:param json_str: json字符串
|
||||
:type json_str: str
|
||||
:return: py对象
|
||||
"""
|
||||
return json.loads(json_str)
|
||||
|
||||
|
||||
def handle_url(protocol, domain, port, path):
|
||||
|
|
|
|||
Loading…
Reference in New Issue