首页 / PYTHON / Yearning sql工单系统 自动执行工单python脚本

Yearning sql工单系统 自动执行工单python脚本

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Yearning sql工单系统 自动执行工单python脚本，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含2082字，纯文字阅读大概需要3分钟。

内容图文

Yearning是一个mysql开源sql语句审核平台，工单流程：开发者提交sql工单->主管审核->运维执行

如果你只是一个工单执行者可以屏蔽一些关键词后自动执行工单

            #
            !/usr/bin/python3
            
#
             -*- coding: utf-8 -*-
            import
             requests

            import
             jsonpath

            import
             time

            import
             sys

            import
             json



            def
             get_token():
    
            #
             获取认证的token
    data = {
        "username": "",
        "password": ""
    }
    headers = {
        "Accept": "application/json"
    }
    request = requests.post(api_url + "ldapauth", data=data, headers=headers)
    request = request.json()
    token = jsonpath.jsonpath(request, "$.token")[0]
    return token


def get_work():
    # 获取工单列表
    headers = {
        "Authorization": "JWT" + "" + token
    }
    request = requests.get(
        api_url + ‘audit_sql?page=1&query={"picker":["",""],"user":"","valve":false}‘,
        headers=headers)
    return request.json()


def get_sql():
    # 获取工单里面的sql
    headers = {
        "Accept": "application/json",
        "Authorization": "JWT" + "" + token
    }
    request = requests.get(
        api_url + ‘getsql?id=‘ + str(id) + ‘&bundle_id=‘ + str(bundle_id),
        headers=headers)
    return request.json()


def audit_sql():
    # 执行工单
    data = {
        "type": 1,
        "to_user": username,
        "id": id
    }
    data = json.dumps(data)
    headers = {
        "Content-Type": "application/json",
        "Authorization": "JWT" + "" + token
    }
    request = requests.put(api_url + ‘audit_sql‘, data=data, headers=headers)
    return request.text


if__name__ == "__main__":
    api_url = "https://yearning.example.com/api/v1/"
    token = get_token()
    work_list = get_work()
    for v in work_list[‘data‘]:
        id = v[‘id‘]
        bundle_id = v[‘bundle_id‘]
        status = v[‘status‘]
        type = v[‘type‘]
        username = v[‘username‘]
        work_id = v[‘work_id‘]
        if status == 2:           # 工单状态等于2则代表待执行
            sql = get_sql()[‘sql‘]
            sql = str.lower(sql)
            keyword = ["drop", "truncate", "grant", "lock"]
            for i in range(0, len(keyword)):
                key = keyword[i]
                if key in sql:
                    print("停止执行有敏感词:" + key)
                    sys.exit()
            audit_sql_return = audit_sql()   #执行工单
            now_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
            log = now_time + ‘‘ + audit_sql_return + ‘‘ + work_id + ‘ "‘ + sql + ‘"‘
            log_file = ‘/tmp/audit-sql.log‘print(log)
            f = open(log_file, ‘a‘)
            f.write(log + ‘\n‘)
            f.close()
            time.sleep(2)

 

原文：https://www.cnblogs.com/37yan/p/12566372.html

内容总结

以上是互联网集市为您收集整理的Yearning sql工单系统 自动执行工单python脚本全部内容，希望文章能够帮你解决Yearning sql工单系统 自动执行工单python脚本所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1096715.html

来源：【匿名】