forked from sheapchen/jf_jd_cfd
-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathql_api.py
100 lines (84 loc) · 2.47 KB
/
ql_api.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
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
import json
import time
import requests
ql_auth_path = '/ql/config/auth.json'
# ql_auth_path = r'D:\Docker\ql\config\auth.json'
ql_url = 'http://localhost:5600'
def __get_token() -> str or None:
with open(ql_auth_path, 'r', encoding='utf-8') as f:
j_data = json.load(f)
return j_data.get('token')
def __get__headers() -> dict:
headers = {
'Accept': 'application/json',
'Content-Type': 'application/json;charset=UTF-8',
'Authorization': 'Bearer ' + __get_token()
}
return headers
# 查询环境变量
def get_envs(name: str = None) -> list:
params = {
't': int(time.time() * 1000)
}
if name is not None:
params['searchValue'] = name
res = requests.get(ql_url + '/api/envs', headers=__get__headers(), params=params)
j_data = res.json()
if j_data['code'] == 200:
return j_data['data']
return []
# 新增环境变量
def post_envs(name: str, value: str, remarks: str = None) -> list:
params = {
't': int(time.time() * 1000)
}
data = [{
'name': name,
'value': value
}]
if remarks is not None:
data[0]['remarks'] = remarks
res = requests.post(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
j_data = res.json()
if j_data['code'] == 200:
return j_data['data']
return []
# 修改环境变量
def put_envs(_id: str, name: str, value: str, remarks: str = None) -> bool:
params = {
't': int(time.time() * 1000)
}
data = {
'name': name,
'value': value,
'_id': _id
}
if remarks is not None:
data['remarks'] = remarks
res = requests.put(ql_url + '/api/envs', headers=__get__headers(), params=params, json=data)
j_data = res.json()
if j_data['code'] == 200:
return True
return False
# 禁用环境变量
def disable_env(_id: str) -> bool:
params = {
't': int(time.time() * 1000)
}
data = [_id]
res = requests.put(ql_url + '/api/envs/disable', headers=__get__headers(), params=params, json=data)
j_data = res.json()
if j_data['code'] == 200:
return True
return False
# 启用环境变量
def enable_env(_id: str) -> bool:
params = {
't': int(time.time() * 1000)
}
data = [_id]
res = requests.put(ql_url + '/api/envs/enable', headers=__get__headers(), params=params, json=data)
j_data = res.json()
if j_data['code'] == 200:
return True
return False