MyRequest.py
5.92 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
#!/usr/bin/python
# -*- coding: UTF-8 -*-
import requests
import json
from commons import common as com
from commons.Logging import Logger
# from requests.adapters import HTTPAdapter HTTPAdapter(max_retries=3)
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
log=Logger()
class myrequest(object):
def __init__(self):
self.timeout = 60
self.max_retries = 3
self.keep_alive = False
self.ssl_verify=False
self.proxies=None
self.allow_redirects=False
def getSession(self,proxies=None):
if proxies:
self.proxies={'http': 'http://localhost:8888', 'https': 'http://localhost:8888'}
se=requests.session()
url=com.get_api_config('login', 'login01', 'url')
header=json.loads(com.get_api_config('login', 'login01', 'header'))
body=com.get_api_config('login', 'login01', 'body')
r=se.post(url,data=body,headers=header,allow_redirects=False,proxies=self.proxies)
# assert r.status_code==200,"请检查登录信息"
print(r.status_code)
self.se=se
return self.se
def get(self,url,**kwargs):
log.info('{0:=^80}'.format('进行GET请求'))
log.info('请求地址为{};\n'.format(url))
res = self.se.get(url, proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
def post(self,url, data=None, json=None, **kwargs):
log.info('{0:=^80}'.format('进行POST请求'))
log.info('请求地址为{};\n 请求body为{};\n'.format(url,data))
res = self.se.post(url, data=data, json=json,proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
def post2(self,api,**kwargs):
self.api=api.split("_")[2]
self.api_num=(api.split("_")[2]+api.split("_")[1])
self.method=com.get_api_config(self.api,self.api_num,'method')
self.body_format=com.get_api_config(self.api,self.api_num,'body_format')
self.url=com.get_api_config(self.api,self.api_num, 'url')
self.data=None if self.body_format=="JSON" else com.get_api_config(self.api,self.api_num,'body')
self.json=None if self.body_format=="FORM" else json.loads(com.get_api_config(self.api,self.api_num,'body'))
self.headers=json.loads(com.get_api_config(self.api,self.api_num,'header'))
log.info('{0:=^80}'.format('请求接口为{}'.format(api)))
log.info('请求地址为{}; 请求方式为{}; 请求body格式为{};\n 请求body为{}{};'.format(
self.url,
self.method,
self.body_format,
self.data,
self.json))
if self.method.upper()=="POST":
res = self.se.post(self.url, data=self.data, json=self.json,proxies=self.proxies,headers=self.headers,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
elif self.method.upper()=="GET":
res = self.se.GET(self.url,proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
elif self.method.upper()=="PUT":
res = self.se.put(self.self.url, data=self.data, json=self.json,proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
elif self.method.upper()=="DELETE":
res = self.se.delete(self.self.url,proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
elif self.method.upper()=="HEAD":
res = self.se.head(self.url,proxies=self.proxies,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
else :
raise Exception("Request method error")
def close_session(self):
self.se.close()
# def to_request(self,method,url,data=None,is_json=False,**kwargs): #最后加入**kwargs使可以传headers
# method = method.upper()
# if isinstance(data,str):
# try:
# data = json.load(data) #'{"mobilephone": "18911114444", "pwd": "123456", "regname": "youxiu"}'
# except Exception as e:
# # print("异常为{}".format(e))
# Logger.error("异常为{}".format(e))
# data = eval(data) #'{"mobilephone": '18911114444',1:"4545"}' 非json的字符串
# if method == 'GET':
# res = self.one_session.request(method,url,params=data,**kwargs)
# elif method == 'POST':
# if is_json:
# res = self.one_session.request(method, url, json=data,**kwargs)
# else:
# res = self.one_session.request(method, url, data=data,**kwargs)
# else:
# res = None
# # print("405 METHOD NOT ALLOWED")
# Logger.error("METHOD NOT ALLOWED".format(method))
# return res
#
# test=myrequest()
# # 获取session
# re=test.getSession(True)
# print(re.headers)
# # 请求
# # re2=test.post(com.get_api_config('upStream', 'upStream01', 'url'),
# # json=json.loads(d),
# # headers=json.loads(com.get_api_config('upStream', 'upStream01', 'header')))
# # print(re2.json())
#
# d=com.get_api_config('upStream', 'upStream01', 'body')
# print(type(json.loads(d)))
# print(d)
# print(d.encode().decode('unicode-escape'))
#
# re3=test.post2("test_01_upStream",body="json")
# print(re3.json())