-
Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathBase_Rest_Class.py
More file actions
130 lines (109 loc) · 4.71 KB
/
Copy pathBase_Rest_Class.py
File metadata and controls
130 lines (109 loc) · 4.71 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
import requests, json, pandas as pd, datetime as dt, sys, urllib,asyncio
from concurrent.futures import ThreadPoolExecutor
import unittest, time, re
###########################################################
######################Parms for Job########################
if len(sys.argv) > 1:
Org = sys.argv[1]
User = sys.argv[2]
Pwd = sys.argv[3]
domain = sys.argv[4]
domain_type = sys.argv[5]
page = sys.argv[6]
size = sys.argv[7]
Poll_Int = sys.argv[8]
else:
Org = ""
User =
Pwd =
domain =
domain_type = "
page = "0"
size = "1000"
Poll_Int = 60
###########################################################
orgs = {}
###########################################################
############################################################
###################### get meta data #######################
class ELN():
def __init__(self, Org:str, User:str, Pwd:str):
self.org = Org
self.User = User
self.Pwd = Pwd
self.domain = domain
self.domain_type = domain_type
self.page = page
self.size = size
self.verifcationErrors = []
def authorize(self)->str:
try:
self.url = ""
self.payload = {"username": f"{User}", "password": f"{Pwd}"}
self.header = {'Content-Type': 'application/json'}
self.response = requests.request("POST", self.url, headers=self.header, data=self.payload)
self.sessionid = self.response.headers[]
return self.sessionid
except Exception as e:
return 'Authentication Failed: ' + str(e)
def get_joburls(self)->dict:
# initial request to get page count and status #
self.iterator = 0
self.url_list = []
self.token_list = []
try:
get_joburls_headers = {
'authorization': f"",
'accept': 'application/json',
'Content-Type': 'application/json'
}
self.job_url = f""
self.job_payload = json.dumps({})
self.job_header = get_joburls_headers
self.metadata = requests.request("POST", self.job_url, headers=self.job_header, data=self.job_payload)
self.totalpages = self.metadata.json()
while self.iterator < self.totalpages:
## scroll required since application is using elastisearch ##
self.page = self.iterator
self.joburl = f""
self.url_list.append(self.joburl)
self.token_list.append(self.authorize())
self.iterator += 1
return dict(zip(self.token_list,self.url_list))
except Exception as e:
return 'Metadata fetch failed, nothing returned from server \n exception: ' + str(e) + "\n" + str(self.metadata.content)
##########################################################
################# retrieve actual data ###################
class get_data():
def __init__(self, urllist:dict,maxpages:int):
self.urllist = urllist
self.maxpages = maxpages
def fetch(self)->list:#,session,url):#,session, id):
tmp =[]
try:
for token,url in self.urllist.items():
get_joburls_headers = {
'authorization': f"",
'accept': 'application/json',
'Content-Type': 'application/json'
,'Connection': 'close'
}
self.job_payload = json.dumps({})
self.job_header = get_joburls_headers
self.jobdata = requests.request("POST", url, headers=self.job_header, data=self.job_payload)
data = self.jobdata.json()
if self.jobdata.status_code != 200:
print("Unable to return data from URL::{0}".format(url),'\n',\
"Failure On Token::{0}".format(token), '\n'\
"Failure Response::{0}".format(self.jobdata.headers)
, json.dumps(self.jobdata.json(),indent=4))
exit(-1)
tmp.append(data)
return tmp
except Exception as e:
return 'fetch data failed, no response from server: ' + str(e) + "\n" + str(
self.jobdata.content)
######################################################################################
#pwd = ELN(Org, User, Pwd).authorize()
#urllist = ELN(Org, User, Pwd).get_joburls()
#data = get_data(urllist= urllist,maxpages=100100).fetch()