|
- # -*- coding: utf-8 -*-
- import time
- from requests.adapters import HTTPAdapter
- import json
- import requests
-
-
- class HttpRequests:
- __slots__ = ("session")
-
- def __init__(self):
- self.session = requests.Session()
- self.session.mount('http://', HTTPAdapter(max_retries=3))
- self.session.mount('https://', HTTPAdapter(max_retries=3))
-
- def send_request(self, method, url, data=None, params_type='JSON', **kwargs):
- method = method.upper()
- params_type = params_type.upper()
- if isinstance(data, str):
- try:
- data = json.loads(data)
- except Exception:
- data = eval(data)
- if 'GET' == method:
- response = self.session.request(method=method, url=url, params=data, **kwargs)
- elif 'POST' == method:
- if params_type == 'FORM': # 发送表单数据，使用data参数传递
- response = self.session.request(method=method, url=url, data=data, **kwargs)
- elif params_type == 'JSON': # 如果接口支持application/json类型，则使用json参数传递
- response = self.session.request(method=method, url=url, json=data, **kwargs)
- else: # 如果接口需要传递其他类型的数据比如 上传文件，调用下面的请求方法
- response = self.session.request(method=method, url=url, **kwargs)
- elif 'PUT' == method:
- if params_type == 'FORM': # 发送表单数据，使用data参数传递
- response = self.session.request(method=method, url=url, data=data, **kwargs)
- elif params_type == 'JSON': # 如果接口支持application/json类型，则使用json参数传递
- response = self.session.request(method=method, url=url, json=data, **kwargs)
- else: # 如果接口需要传递其他类型的数据比如 上传文件，调用下面的请求方法
- response = self.session.request(method=method, url=url, **kwargs)
- elif 'DELETE' == method:
- if params_type == 'FORM': # 发送表单数据，使用data参数传递
- response = self.session.request(method=method, url=url, data=data, **kwargs)
- elif params_type == 'JSON': # 如果接口支持application/json类型，则使用json参数传递
- response = self.session.request(method=method, url=url, json=data, **kwargs)
- else: # 如果接口需要传递其他类型的数据比如 上传文件，调用下面的请求方法
- response = self.session.request(method=method, url=url, **kwargs)
- # 如果请求方式非 get 和post 会报错，当然你也可以继续添加其他的请求方法
- else:
- raise ValueError('request method "{}" error ! please check'.format(method))
-
- return response
-
- def __call__(self, method, url, data=None, params_type='JSON', **kwargs):
- return self.send_request(method, url,
- data=data,
- params_type=params_type,
- **kwargs)
-
- def close_session(self):
- self.session.close()
- try:
- del self.session.cookies['JSESSIONID']
- except:
- pass
-
- def __del__(self):
- self.close_session()
-
- if __name__ == '__main__':
- start_time = time.time()
- headers = {'content-type': "application/json"}
- req = HttpRequests()
- response1 = req.send_request('GET', "http://localhost:19610/api/v1/getChannelList?offset=0&row=1", headers=headers, timeout=3)
- result = response1.content.decode('utf-8')
- print(result)
- result = json.loads(result)
- result = result["ChannelList"]
- print(result[0]["enable"], type(result[0]["enable"]))
- result[0]["enable"] = "true"
- result[0]["srcURL"] = '"%s"' % result[0]["srcURL"]
- param = 'indexCode=%s&name=%s&srcURL=%s&connectType=%s&timeout=5&mediaType=%s&dstURL=%s&dstFormat=%s&enable=%s' %\
- (result[0]["indexcode"], result[0]["name"], result[0]["srcURL"], result[0]["connectType"],
- result[0]["mediaType"], result[0]["dstURL"], result[0]["dstFormat"],result[0]["enable"])
- url = 'http://localhost:19610/api/v1/updateChannel?%s' % param
- response = req.send_request('GET', url, headers=headers, timeout=10)
- print(response.__dict__)
- print(response.status_code)
- print(type(response.content), isinstance(response.content, bytes))
- print(response.content.decode('utf-8'), type(response.content.decode('utf-8')))
- req.close_session()
- print(time.time() - start_time)
|
