数据库审核工具SQLE接口调用
接上文数据库审核接口SQLE的探索使用，本次自定义接口进行调用，实现需求。
1、创建自定义审核接口
因直接调用SQLE的审核接口，会出现token过期，且审核及结果查询接口是分开的，因此，出于以上2个原因的考虑，重新编写一个接口用于调用SQLE的审核接口并返回审核结果。参考代码如下：
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
|
# coding=utf-8
import requests
from flask import request
import json
from flask import Flask
app = Flask(__name__)
@app.route('/audit_task', methods=['post','get'])
def audit_task():
get_data = request.args.to_dict()
v_instance = get_data.get('instance')
v_schema = get_data.get('schema')
v_sql = get_data.get('sql')
v_token = get_token()
v_headers = {'Authorization': f"{v_token}"}
#print(v_headers)
url = 'http://127.0.0.1:10000/v1/tasks/audits'
v_data_a = {'instance_name':v_instance,'instance_schema':v_schema,'sql':v_sql}
#print(v_data_a)
res = requests.post(url, data=v_data_a, headers=v_headers)
v_taskid=res.json()['data']['task_id']
url2='http://127.0.0.1:10000/v1/tasks/audits/'+str(v_taskid)+'{/sqls?page_index=1&page_size=10&no_duplicate=false'
# print(url2)
res2 = requests.get(url2, headers=v_headers)
audit_result = res2.json()['data'][0]['audit_result']
if not(audit_result):
audit_result=f'审核通过'
return audit_result
def get_token():
url = 'http://127.0.0.1:10000/v1/login'
data = {"username": "admin", "password": "xxxx"}
# res = requests.post(url, json=data)
h = {
'Content-Type': 'application/json; charset=UTF-8'
}
res = requests.post(url, data=json.dumps(data), headers=h)
t = json.loads(res.text)
#print(type(json.loads(res.text)))
token = t['data']['token']
#print(t['data']['token'])
return token
if __name__ == '__main__':
app.run(host='0.0.0.0',port=5000, debug=True)
|
注：使用时，需提前在SQLE中配置好实例信息。SQLE部署方式及简单配置可参考
数据库审核工具SQLE部署及使用
2. 调用测试
运行以上接口，并使用postman调用进行测试
输入参数名特地调整了一下，以便于区别，测试情况如下
调用结果如下：
1
2
3
4
|
[error]除了自增列及大字段列之外，每个列都必须添加默认值
[error]表必须有主键
[error]新建表必须加入if not exists create，保证重复执行不报错
|
与上一篇结果一致，且 无需调用2次接口。
