📢 转载信息
原文链接：https://www.kdnuggets.com/5-useful-diy-python-functions-for-json-parsing-and-processing
原文作者：Bala Priya C
Image by Author
#
引言
在Python中处理JSON通常充满挑战。基本的json.loads()只能满足基本需求。
API响应、配置文件和数据导出中经常包含杂乱或结构不良的JSON。你需要扁平化嵌套对象、安全地提取值以避免KeyError异常、合并多个JSON文件或在JSON与其他格式之间进行转换。这些任务在网络抓取、API集成和数据处理中经常出现。本文将介绍五个实用的函数，用于处理常见的JSON解析和处理任务。
你可以在GitHub上找到这些函数的代码。
#
1. 安全地提取嵌套值
JSON对象通常有多个嵌套层级。使用方括号表示法访问深度嵌套的值会很快变得棘手。如果缺少任何键，你将得到一个KeyError。
下面是一个函数，它允许你使用点表示法访问嵌套值，并在键丢失时提供回退值：
def get_nested_value(data, path, default=None): """ 使用点表示法安全地从JSON中提取嵌套值。 参数: data: 字典或JSON对象 path: 点分隔的字符串，如 "user.profile.email" default: 如果路径不存在时返回的值 返回: 路径处的值，如果找不到则返回默认值 """ keys = path.split('.') current = data for key in keys: if isinstance(current, dict): current = current.get(key) if current is None: return default elif isinstance(current, list): try: index = int(key) current = current[index] except (ValueError, IndexError): return default else: return default return current
让我们用一个复杂的嵌套结构来测试它：
# 示例JSON数据 user_data = { "user": { "id": 123, "profile": { "name": "Allie", "email": "allie@example.com", "settings": { "theme": "dark", "notifications": True } }, "posts": [ {"id": 1, "title": "First Post"}, {"id": 2, "title": "Second Post"} ] } } # 提取值 email = get_nested_value(user_data, "user.profile.email") theme = get_nested_value(user_data, "user.profile.settings.theme") first_post = get_nested_value(user_data, "user.posts.0.title") missing = get_nested_value(user_data, "user.profile.age", default=25) print(f"Email: {email}") print(f"Theme: {theme}") print(f"First post: {first_post}") print(f"Age (default): {missing}")
输出：
Email: allie@example.com Theme: dark First post: First Post Age (default): 25
该函数以点为分隔符拆分路径字符串，并逐个键遍历数据结构。在每个级别，它会检查当前值是字典还是列表。对于字典，它使用.get(key)，它会返回None而不是引发错误来处理缺失的键。对于列表，它尝试将键转换为整数索引。
default参数在路径的任何部分不存在时提供回退值。这可以防止你在处理来自API的不完整或不一致的JSON数据时代码崩溃。
当处理可选字段或仅在特定条件下存在的字段时，这种模式特别有用。
#
2. 将嵌套JSON扁平化为单层字典
机器学习模型、CSV导出和数据库插入通常需要扁平的数据结构。但API响应和配置文件使用嵌套的JSON。将嵌套对象转换为扁平的键值对是一项常见任务。
下面是一个使用可自定义分隔符扁平化嵌套JSON的函数：
def flatten_json(data, parent_key='', separator='_'): """ 将嵌套JSON扁平化为单层字典。 参数: data: 嵌套字典或JSON对象 parent_key: 键的前缀（用于递归） separator: 连接嵌套键的字符串 返回: 带有连接键的扁平化字典 """ items = [] if isinstance(data, dict): for key, value in data.items(): new_key = f"{parent_key}{separator}{key}" if parent_key else key if isinstance(value, dict): # 递归扁平化嵌套字典 items.extend(flatten_json(value, new_key, separator).items()) elif isinstance(value, list): # 使用索引键扁平化列表 for i, item in enumerate(value): list_key = f"{new_key}{separator}{i}" if isinstance(item, (dict, list)): items.extend(flatten_json(item, list_key, separator).items()) else: items.append((list_key, item)) else: items.append((new_key, value)) else: items.append((parent_key, data)) return dict(items)
现在让我们扁平化一个复杂的嵌套结构：
# 复杂嵌套JSON product_data = { "product": { "id": 456, "name": "Laptop", "specs": { "cpu": "Intel i7", "ram": "16GB", "storage": { "type": "SSD", "capacity": "512GB" } }, "reviews": [ {"rating": 5, "comment": "Excellent"}, {"rating": 4, "comment": "Good value"} ] } } flattened = flatten_json(product_data) for key, value in flattened.items(): print(f"{key}: {value}")
输出：
product_id: 456 product_name: Laptop product_specs_cpu: Intel i7 product_specs_ram: 16GB product_specs_storage_type: SSD product_specs_storage_capacity: 512GB product_reviews_0_rating: 5 product_reviews_0_comment: Excellent product_reviews_1_rating: 4 product_reviews_1_comment: Good value
该函数使用递归来处理任意的嵌套深度。当遇到字典时，它会处理每个键值对，通过连接父键和分隔符来构建扁平化的键。
对于列表，它使用索引作为键的一部分。这使得你可以在扁平化的输出中保留数组元素的顺序和结构。reviews_0_rating模式告诉你这是来自第一个评论的评分。
separator参数允许你自定义输出格式。根据需要，可以使用点表示法、下划线（snake_case）或斜杠（路径状键）。
当需要将JSON API响应转换为需要唯一列名的DataFrame或CSV行时，此函数特别有用。
#
3. 深度合并多个JSON对象
配置管理通常需要合并包含默认设置、特定于环境的配置、用户偏好等的多个JSON文件。简单的dict.update()只处理顶层。你需要深度合并，它能递归地组合嵌套结构。
下面是一个深度合并JSON对象的函数：
def deep_merge_json(base, override): """ 深度合并两个JSON对象，覆盖项优先。 参数: base: 基础字典 override: 包含要覆盖/添加的值的字典 返回: 合并后的新字典 """ result = base.copy() for key, value in override.items(): if key in result and isinstance(result[key], dict) and isinstance(value, dict): # 递归合并嵌套字典 result[key] = deep_merge_json(result[key], value) else: # 覆盖或添加值 result[key] = value return result
让我们尝试合并示例配置信息：
import json # 默认配置 default_config = { "database": { "host": "localhost", "port": 5432, "timeout": 30, "pool": { "min": 2, "max": 10 } }, "cache": { "enabled": True, "ttl": 300 }, "logging": { "level": "INFO" } } # 生产环境覆盖配置 prod_config = { "database": { "host": "prod-db.example.com", "pool": { "min": 5, "max": 50 } }, "cache": { "ttl": 600 }, "monitoring": { "enabled": True } } merged = deep_merge_json(default_config, prod_config) print(json.dumps(merged, indent=2))
输出：
{ "database": { "host": "prod-db.example.com", "port": 5432, "timeout": 30, "pool": { "min": 5, "max": 50 } }, "cache": { "enabled": true, "ttl": 600 }, "logging": { "level": "INFO" }, "monitoring": { "enabled": true } }
该函数递归地合并嵌套的字典。当基础和覆盖项在同一键下都包含字典时，它会合并这些字典而不是完全替换它们。这样可以保留未显式覆盖的值。
注意database.port和database.timeout仍保留自默认配置，而database.host被覆盖。pool设置在嵌套级别合并，因此min和max都得到了更新。
该函数还会添加基础配置中不存在的新键，如生产覆盖中的monitoring部分。
你可以通过链式调用多个合并操作来叠加配置：
final_config = deep_merge_json( deep_merge_json(default_config, prod_config), user_preferences )
这种模式在应用程序配置中很常见，其中包含默认值、特定于环境的设置和运行时覆盖。
#
4. 根据Schema或白名单过滤JSON
API通常返回比你需要的数据多得多的信息。大型JSON响应会使代码难以阅读。有时你只需要特定的字段，或者在记录日志之前需要删除敏感数据。
下面是一个仅保留指定字段的JSON过滤函数：
def filter_json(data, schema): """ 根据Schema过滤JSON，仅保留Schema中指定的字段。 参数: data: 要过滤的字典或JSON对象 schema: 定义要保留哪些字段的字典 使用True保留字段，使用嵌套的dict进行嵌套过滤 返回: 仅包含指定字段的过滤后的字典 """ if not isinstance(data, dict) or not isinstance(schema, dict): return data result = {} for key, value in schema.items(): if key not in data: continue if value is True: # 原样保留此字段 result[key] = data[key] elif isinstance(value, dict): # 递归过滤嵌套对象 if isinstance(data[key], dict): filtered_nested = filter_json(data[key], value) if filtered_nested: result[key] = filtered_nested elif isinstance(data[key], list): # 过滤列表中的每个项 filtered_list = [] for item in data[key]: if isinstance(item, dict): filtered_item = filter_json(item, value) if filtered_item: filtered_list.append(filtered_item) else: filtered_list.append(item) if filtered_list: result[key] = filtered_list return result
让我们过滤一个示例API响应：
import json # 示例API响应 api_response = { "user": { "id": 789, "username": "Cayla", "email": "cayla@example.com", "password_hash": "secret123", "profile": { "name": "Cayla Smith", "bio": "Software developer", "avatar_url": "https://example.com/avatar.jpg", "private_notes": "Internal notes" }, "posts": [ { "id": 1, "title": "Hello World", "content": "My first post", "views": 100, "internal_score": 0.85 }, { "id": 2, "title": "Python Tips", "content": "Some tips", "views": 250, "internal_score": 0.92 } ] }, "metadata": { "request_id": "abc123", "server": "web-01" } } # 定义要保留内容的Schema public_schema = { "user": { "id": True, "username": True, "profile": { "name": True, "avatar_url": True }, "posts": { "id": True, "title": True, "views": True } } } filtered = filter_json(api_response, public_schema) print(json.dumps(filtered, indent=2))
输出：
{ "user": { "id": 789, "username": "Cayla", "profile": { "name": "Cayla Smith", "avatar_url": "https://example.com/avatar.jpg" }, "posts": [ { "id": 1, "title": "Hello World", "views": 100 }, { "id": 2, "title": "Python Tips", "views": 250 } ] } }
Schema充当白名单。将其设置为True会将其包含在输出中。使用嵌套字典可以过滤嵌套对象。该函数递归地将Schema应用于嵌套结构。
对于数组，Schema应用于每个项。在示例中，posts数组被过滤，因此每个帖子只包含id、title和views，而content和internal_score被排除。
请注意，像password_hash和private_notes这样的敏感字段不会出现在输出中。这使得该函数在记录日志或将数据发送到前端应用程序之前进行数据清理时非常有用。
你可以为不同的用例创建不同的Schema，例如列表视图的最小Schema、单项视图的详细Schema以及包含所有内容的管理员Schema。
#
5. 将JSON转换为点表示法及反之
有些系统使用扁平的键值存储，但你希望在代码中使用嵌套的JSON。在扁平的点表示法键和嵌套结构之间进行转换有助于实现这一点。
下面是一对用于双向转换的函数。
//
将JSON转换为点表示法
def json_to_dot_notation(data, parent_key=''): """ 将嵌套JSON转换为扁平的点表示法字典。 参数: data: 嵌套字典 parent_key: 键的前缀（用于递归） 返回: 带有点表示法键的扁平字典 """ items = {} if isinstance(data, dict): for key, value in data.items(): new_key = f"{parent_key}.{key}" if parent_key else key if isinstance(value, dict): items.update(json_to_dot_notation(value, new_key)) else: items[new_key] = value else: items[parent_key] = data return items
//
将点表示法转换为JSON
def dot_notation_to_json(flat_data): """ 将扁平的点表示法字典转换为嵌套的JSON。 参数: flat_data: 带有点表示法键的字典 返回: 嵌套字典 """ result = {} for key, value in flat_data.items(): parts = key.split('.') current = result for i, part in enumerate(parts[:-1]): if part not in current: current[part] = {} current = current[part] current[parts[-1]] = value return result
让我们测试往返转换：
import json # 原始嵌套JSON config = { "app": { "name": "MyApp", "version": "1.0.0" }, "database": { "host": "localhost", "credentials": { "username": "admin", "password": "secret" } }, "features": { "analytics": True, "notifications": False } } # 转换为点表示法（用于环境变量） flat = json_to_dot_notation(config) print("Flat format:") for key, value in flat.items(): print(f" {key} = {value}") print("\n" + "="*50 + "\n") # 转换回嵌套JSON nested = dot_notation_to_json(flat) print("Nested format:") print(json.dumps(nested, indent=2))
输出：
Flat format: app.name = MyApp app.version = 1.0.0 database.host = localhost database.credentials.username = admin database.credentials.password = secret features.analytics = True features.notifications = False ================================================== Nested format: { "app": { "name": "MyApp", "version": "1.0.0" }, "database": { "host": "localhost", "credentials": { "username": "admin", "password": "secret" } }, "features": { "analytics": true, "notifications": false } }
json_to_dot_notation函数通过递归遍历嵌套字典并将键用点连接起来，从而扁平化结构。与前面介绍的扁平化函数不同，这个函数不处理数组；它专门针对纯键值对的配置数据。
dot_notation_to_json函数则反向操作。它对每个键进行点拆分，并在需要时创建中间字典来构建嵌套结构。循环处理除最后一个部分外的所有部分，创建嵌套级别。然后将值赋给最终的键。
这种方法可以在使用扁平键值存储系统的同时，保持配置的可读性和可维护性。
#
总结
JSON处理超出了基本的json.loads()范畴。在大多数项目中，你需要工具来导航嵌套结构、转换形状、合并配置、过滤字段以及在不同格式之间进行转换。
本文中的技术也可以转移到其他数据处理任务中。你可以修改这些模式来处理XML、YAML或自定义数据格式。
从安全访问函数开始，以防止代码中出现KeyError异常。当你遇到特定需求时，再添加其他函数。祝你编码愉快！
Bala Priya C 是一位来自印度的开发者和技术作家。她喜欢在数学、编程、数据科学和内容创作的交叉领域工作。她的兴趣和专业领域包括DevOps、数据科学和自然语言处理。她喜欢阅读、写作、编码和咖啡！目前，她正致力于通过撰写教程、操作指南、观点文章等方式学习并与开发者社区分享她的知识。Bala还创建引人入胜的资源概览和编码教程。
评论区