如何使用Python列出JSON文件所有键以确定SQLite数据库列名
Python读取JSON文件自动提取键导入SQLite实现方案
完全基于Python标准库、JSON、SQLite实现,无额外依赖。
核心逻辑
- 直接使用
json.load()方法读取JSON文件对象,不需要先将文件内容读取为字符串再解析,匹配处理JSON文件而非JSON字符串的要求。 - 遍历解析后的JSON结构,用集合自动收集所有出现过的键,自动去重,不需要手动整理列清单,实现类似元数据自动提取的效果。
- 基于收集到的键自动生成SQLite建表、插入语句,完成数据导入。
完整实现代码
import json import sqlite3 from typing import Set, Tuple def extract_json_metadata(file_path: str) -> Tuple[Set[str], list | dict]: """ 直接从JSON文件提取所有顶层键,同时返回解析后的数据 兼容顶层为单个字典/字典列表两种常见JSON结构 """ all_keys = set() with open(file_path, 'r', encoding='utf-8') as f: # json.load直接接收文件句柄,无需手动处理JSON字符串 parsed_data = json.load(f) # 收集所有存在的键 if isinstance(parsed_data, dict): all_keys.update(parsed_data.keys()) elif isinstance(parsed_data, list): for item in parsed_data: if isinstance(item, dict): all_keys.update(item.keys()) return all_keys, parsed_data def import_json_to_sqlite(json_file_path: str, sqlite_db_path: str, table_name: str = "json_import") -> None: keys, data = extract_json_metadata(json_file_path) # 对列名排序保证表结构稳定 columns = sorted(keys) # 连接SQLite数据库 conn = sqlite3.connect(sqlite_db_path) cursor = conn.cursor() # 自动生成建表语句,默认使用TEXT类型兼容所有JSON基础类型,可按需调整类型映射 create_table_clause = f""" CREATE TABLE IF NOT EXISTS {table_name} ( id INTEGER PRIMARY KEY AUTOINCREMENT, {', '.join([f"`{col}` TEXT" for col in columns])} ) """ cursor.execute(create_table_clause) # 生成批量插入语句 insert_clause = f"INSERT INTO {table_name} ({', '.join([f'`{col}`' for col in columns])}) VALUES ({', '.join(['?' for _ in columns])})" # 整理待插入的数据行 if isinstance(data, dict): rows = [tuple(data.get(col) for col in columns)] elif isinstance(data, list): rows = [tuple(item.get(col) for col in columns) for item in data if isinstance(item, dict)] cursor.executemany(insert_clause, rows) conn.commit() conn.close() print(f"导入完成,自动识别列:{columns}") # 调用示例 if __name__ == "__main__": import_json_to_sqlite("target.json", "result.db")
扩展说明
- 上述代码无任何第三方依赖,完全符合限定的技术栈要求。自动提取键的逻辑不需要人工介入,只要JSON是扁平字典结构,就能自动识别全量字段,哪怕不同记录存在字段缺失也能正常收集所有出现过的键,和提到的
INFORMATION_SCHEMA.COLUMNS、XMLFINDALL的自动元数据提取效果一致。 - 如果需要处理嵌套JSON结构,可以使用递归逻辑收集全层级的键,参考代码片段:
def collect_nested_keys(obj: any, key_prefix: str = "") -> Set[str]: """递归收集嵌套结构的所有键,用.连接层级,例如user.name""" collected = set() if isinstance(obj, dict): for k, v in obj.items(): current_key = f"{key_prefix}.{k}" if key_prefix else k collected.add(current_key) collected.update(collect_nested_keys(v, current_key)) elif isinstance(obj, list): for item in obj: collected.update(collect_nested_keys(item, key_prefix)) return collected
收集到嵌套键后,可以将嵌套值扁平化后导入,也可以直接将嵌套片段序列化为JSON字符串存入TEXT类型字段。
内容的提问来源于stack exchange,提问作者jcrom77
相关产品推荐
相关产品推荐

