一、前言

项目迭代了很多年，使用了存储，存储行数很多，而且涉及到存储的嵌套，为了方便查询问题需要查询存储所依赖的表。以JSON格式的数据输出到文件，方便搜索。

二、源码

import datetime
import json

import cx_Oracle

"""递归查询oracle存储关联的存储和表"""

user = "用户名"
passwd = "密码"
listener = '监听地址'
conn = cx_Oracle.connect(user, passwd, listener)
cursor = conn.cursor()


def select(sql) -> json:
    rows = []
    cursor.execute(sql)
    result = cursor.fetchall()
    col_name = cursor.description
    for row in result:
        d = {}
        for col in range(len(col_name)):
            key = col_name[col][0]
            value = row[col]
            if isinstance(value, datetime.datetime):
                value = value.strftime('%Y-%m-%d %H:%M:%S')
            d[key] = value
        rows.append(d)
    js = json.dumps(rows, sort_keys=True, ensure_ascii=False, separators=(',', ':'))
    return js


# 关闭连接，释放资源
def disconnect():
    cursor.close()
    conn.close()
    print("关闭数据库连接")


def find_dep(table_name):
    sql = "SELECT * FROM user_dependencies t WHERE t.name = '" + table_name + "' order by t.REFERENCED_TYPE desc"
    field_list = select(sql)
    field_list = json.loads(field_list)
    for f in field_list:
        referenced_type = f['REFERENCED_TYPE']
        if referenced_type == 'PROCEDURE':
            # 递归查询
            rs = find_dep(f['REFERENCED_NAME'])
            f['dependencies'] = json.loads(rs)

    return json.dumps(field_list)


if __name__ == '__main__':
    print('开始查询')
    t = 'FR_SW_JOB_CHECK_LIST'
    out_file = open("C:/Users/*/Desktop/" + t + ".json", "w")
    try:
        r = find_dep(t)
        print(r)
        # 将json数据写到文件
        json.dump(json.loads(r), out_file, indent=6)
    finally:
        out_file.close()
        disconnect()