MyRequest.py
4.38 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
#!/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 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,**kwargs)
log.info('响应body为{};'.format(res.json()))
return res
def post2(self,session,api,**kwargs):
'''处理静态数据的方法'''
self.se=session
self.api=api.split("_")[2]
self.api_num=(api.split("_")[2]+api.split("_")[1])
#获取接口的请求方法，url,header,body_format,body等数据
self.method=com.get_api_config(self.api,self.api,'method')
self.body_format=com.get_api_config(self.api,self.api,'body_format')
self.url=com.get_api_config(self.api,self.api, 'url')
self.headers=json.loads(com.get_api_config(self.api,self.api,'header'))
#判断请求数据类型
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'))
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,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()
#
# 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())
# a="test_04_upStream"
# print(a.split("_")[2])
# print(a.split("_")[2]+a.split("_")[1])