-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
63 lines (58 loc) · 2.1 KB
/
utils.py
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
import requests, time, os, inquirer
# COLORS
def red(str): return "\033[91m{}\033[0m".format(str)
def green(str): return "\033[92m{}\033[0m".format(str)
def yellow(str): return "\033[93m{}\033[0m".format(str)
def blue(str): return "\033[94m{}\033[0m".format(str)
def pink(str): return "\033[95m{}\033[0m".format(str)
url = "https://api.intra.42.fr"
errors = []
client = os.environ.get('API42_CLIENT')
if client == None:
errors.append("env API42_CLIENT not set :(")
secret = os.environ.get('API42_SECRET')
if secret == None:
errors.append("env API42_SECRET not set :(")
if len(errors) > 0:
for error in errors:
print(red(error))
exit(1)
access_token = requests.post("{}/oauth/token?grant_type=client_credentials&client_id={}&client_secret={}".format(url, client, secret)).json()["access_token"]
def req_42api(path, payload=dict(), size=100, nb_result=0, number=1):
global access_token
loop = 1
datas = []
internal_payload = {
'page':
{
'number': number,
'size': size
}
}
calc_payload = {**payload, **internal_payload}
while loop == 1:
req = requests.get(url+path, json=calc_payload, headers={"Authorization": "Bearer "+access_token})
if req.status_code == 429:
if float(req.headers['Retry-After']) > 10:
print('Waiting {}s'.format(req.headers['Retry-After']))
time.sleep(float(req.headers["Retry-After"]))
elif req.status_code == 200:
datas += req.json()
if (nb_result == -1 or
int(req.headers["X-Page"]) * int(req.headers["X-Per-Page"]) >= int(req.headers["X-Total"]) or
(nb_result > 0 and int(req.headers["X-Page"]) * int(req.headers["X-Per-Page"]) >= nb_result)):
loop = 0
calc_payload['page']['number'] += 1
elif req.status_code == 401:
access_token = requests.post("{}/oauth/token?grant_type=client_credentials&client_id={}&client_secret={}".format(url, client, secret)).json()["access_token"]
elif req.status_code == 404:
return False
return datas
def print_inquirer(message="Select a choice", choices=[]):
questions = [
inquirer.List('action',
message=message,
choices=choices,
),
]
return inquirer.prompt(questions, raise_keyboard_interrupt=True)['action']