Python中如何从JSON Schema提取表名及对应列名

核心判定规则 #

完全匹配需求定义的提取逻辑：

• 表名判定：字段类型为对象数组（即type: "array"，且数组内元素items的类型为object）时，该字段名即为表名
• 列名判定：表节点下的基础类型属性（非嵌套对象、非对象数组的属性）即为对应表的列名

实现思路 #

针对Schema多层嵌套的结构特点，采用逐节点递归扫描的逻辑：

1. 从Schema最外层根节点开始，逐个遍历节点下的所有属性
2. 遇到符合表判定规则的字段时，自动记录表名，同时扫描表节点下的基础属性作为列名
3. 无论当前字段是不是表节点，都继续向内遍历嵌套结构，直到扫描完全部层级，一次性提取所有嵌套的表信息

可直接运行的Python代码 #

import json

def extract_tables_from_json_schema(schema: dict) -> dict:
    # 存储提取结果：key为表名，value为对应列名列表
    tables = {}

    # 基础列判定规则：不属于对象、数组的字段类型，均为可提取的列
    def is_basic_column(field_config: dict) -> bool:
        field_type = field_config.get("type")
        return field_type not in ["object", "array"]

    def traverse(current_schema: dict):
        # 跳过没有properties结构的无效节点
        if "properties" not in current_schema:
            return
        # 遍历当前节点下的所有字段
        for field_name, field_config in current_schema["properties"].items():
            # 判定当前字段是否为对象数组（表的判定条件）
            is_object_array = (
                field_config.get("type") == "array"
                and field_config.get("items", {}).get("type") == "object"
            )
            if is_object_array:
                tables[field_name] = []
                table_item_schema = field_config["items"]
                # 提取当前表下的所有列
                if "properties" in table_item_schema:
                    for col_name, col_config in table_item_schema["properties"].items():
                        if is_basic_column(col_config):
                            tables[field_name].append(col_name)
                # 向内遍历表内部的嵌套结构，提取下层表
                traverse(table_item_schema)
            # 向内遍历当前字段的嵌套结构，避免漏检表
            traverse(field_config)

    traverse(schema)
    return tables

if __name__ == "__main__":
    # 替换为本地大JSON Schema的实际文件路径
    with open("your_schema_file.json", "r", encoding="utf-8") as f:
        full_schema = json.load(f)
    extraction_result = extract_tables_from_json_schema(full_schema)
    # 结果可直接打印、导出为JSON/CSV用于业务核对
    print(json.dumps(extraction_result, indent=2, ensure_ascii=False))

适配大规模Schema的注意点 #

• 如果存在跨层级重名的表名，可把表名拼接为完整路径（如根节点.父节点.表名）作为结果的key，避免同名覆盖
• 针对600张表的规模，递归遍历的性能完全足够；如果Schema嵌套层级超过Python默认递归深度（1000层），可把递归逻辑替换为栈实现的迭代版本，避免栈溢出
• 基础列的判定规则可根据业务实际调整，比如需要把单值对象类型也纳入列范围时，仅需修改is_basic_column的判定条件即可

示例验证 #

用题目给出的示例Schema跑通上述代码后，输出结果完全匹配预期：

{
  "CLASS_B": ["NAME1", "NAME2"],
  "CLASS_C": ["NAME3", "NAME4"]
}

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