如何用Python解析Oracle表DDL并生成指定格式的JSON数据？

问题 #

需要解析Oracle表的DDL并生成指定嵌套结构的JSON数据，使用Python的ddlparse库编写的代码目前只能输出单列的JSON字符串，无法达到目标格式要求。

原代码 #

import json

from ddlparse import DdlParse

sample_ddl = """
CREATE TABLE SCOTT.EMPLOYEE (   
EMP_NO VARCHAR2(5 CHAR) NOT NULL , 
    EMP_NAME VARCHAR2(50 CHAR) NOT NULL , 
    CONSTRAINT EMP_PK PRIMARY KEY (EMP_NO)
);
"""

# parse pattern (1-2) : Specify source database
table = DdlParse().parse(ddl=sample_ddl, source_database=DdlParse.DATABASE.oracle)


print("* COLUMN *")
for col in table.columns.values():
     
    col_info = {}

    #col_info["table:name"]            = table.name
    col_info["name"]                  = col.name
    col_info["data_type"]             = col.data_type
    col_info["length"]                = col.length
    col_info["precision(=length)"]    = col.precision
    col_info["scale"]                 = col.scale
    col_info["not_null"]              = col.not_null
    col_info["PK"]                    = col.primary_key
    col_info["bigquery_field"]        = json.loads(col.to_bigquery_field())   

    print(json.dumps(col_info, indent=2, ensure_ascii=False))

期望输出格式 #

{
    "name": "EMP",
    "columns":
    [
        {
            "name": "EMP_NO",
            "dataType": "VARCHAR2(5 CHAR)",
            "nullable": false
        },
        {
            "name": "EMP_NAME",
            "dataType": "VARCHAR2(50 CHAR)",
            "nullable": false
        }
    ],
    "primaryKey":
    [
        "EMP_NO"
    ]
}

修改后的代码 #

import json
from ddlparse import DdlParse

sample_ddl = """
CREATE TABLE SCOTT.EMPLOYEE (   
EMP_NO VARCHAR2(5 CHAR) NOT NULL , 
    EMP_NAME VARCHAR2(50 CHAR) NOT NULL , 
    CONSTRAINT EMP_PK PRIMARY KEY (EMP_NO)
);
"""

# 解析Oracle DDL
table = DdlParse().parse(ddl=sample_ddl, source_database=DdlParse.DATABASE.oracle)

# 构建目标JSON结构框架
result = {
    "name": "EMP",  # 若需使用DDL中的表名，替换为 table.name.split(".")[-1]
    "columns": [],
    "primaryKey": []
}

# 填充列信息
for col in table.columns.values():
    column_item = {
        "name": col.name,
        "dataType": f"{col.data_type}({col.length} CHAR)",
        "nullable": not col.not_null
    }
    result["columns"].append(column_item)

# 填充主键信息
if table.primary_key:
    result["primaryKey"] = [col.name for col in table.primary_key.columns]

# 输出格式化后的完整JSON
print(json.dumps(result, indent=2, ensure_ascii=False))

代码说明 #

1. 结构初始化：先创建包含name、columns、primaryKey的顶层字典，作为最终JSON的骨架。
2. 列数据处理：遍历解析得到的列对象，拼接出dataType的完整格式，将not_null取反得到nullable字段值，逐个添加到columns列表中。
3. 主键提取：从解析结果的primary_key属性中提取所有主键列名，组成列表赋值给primaryKey。
4. 统一输出：用json.dumps一次性生成完整的嵌套结构JSON，替代原代码逐行输出单列数据的方式。

内容的提问来源于stack exchange，提问作者Amdone