如何在Python中统一requests响应的字典键命名格式?
Pythonic 高效整合异构JSON格式方案
一、延迟转换:用动态访问类封装数据
核心思路是不提前全量遍历转换,仅在实际访问键时实时处理格式转换,适合大数据量场景,避免一次性内存开销,尤其适配Lambda这类资源受限环境。
可以封装一个UnifiedData类,重载键访问和属性访问方法,自动处理蛇形/驼峰键的转换:
import re from typing import Any, Dict, List def snake_to_camel(s: str) -> str: parts = s.split('_') return parts[0] + ''.join(p.capitalize() for p in parts[1:]) class UnifiedData: def __init__(self, data: Dict[str, Any], is_snake_case: bool): self.raw_data = data self.is_snake = is_snake_case def _convert_key(self, key: str) -> str: return snake_to_camel(key) if self.is_snake else key def __getitem__(self, key: str) -> Any: target_key = self._convert_key(key) if self.is_snake else key value = self.raw_data.get(target_key) # 仅在访问嵌套结构时递归封装 if isinstance(value, dict): return UnifiedData(value, self.is_snake) elif isinstance(value, list): return [UnifiedData(item, self.is_snake) if isinstance(item, dict) else item for item in value] return value # 支持属性访问,如 data.userName 替代 data['userName'] def __getattr__(self, attr: str) -> Any: return self.__getitem__(attr)
使用示例:
# 服务A的蛇形数据 service_a_data = {"user_id": 123, "user_name": "Alice", "order_details": {"order_id": 456, "total_amount": 99.9}} # 服务B的驼峰数据 service_b_data = {"userId": 456, "userName": "Bob", "orderDetails": {"orderId": 789, "totalAmount": 199.9}} # 统一封装为驼峰访问格式 unified_a = UnifiedData(service_a_data, is_snake_case=True) unified_b = UnifiedData(service_b_data, is_snake_case=False) # 用统一的驼峰键访问 print(unified_a.userName) # 输出 Alice print(unified_b.userName) # 输出 Bob print(unified_a.orderDetails.totalAmount) # 输出 99.9
二、请求侧即时转换:结合requests响应钩子
如果不需要延迟处理,希望拿到响应后直接得到统一格式数据,可以用requests的响应钩子,在解析JSON时同步转换键名:
import requests from typing import Any, Dict def snake_to_camel(s: str) -> str: parts = s.split('_') return parts[0] + ''.join(p.capitalize() for p in parts[1:]) def convert_snake_response(response: requests.Response, *args, **kwargs) -> None: if response.headers.get('Content-Type') == 'application/json': def _recursive_convert(data: Any) -> Any: if isinstance(data, dict): return {snake_to_camel(k): _recursive_convert(v) for k, v in data.items()} elif isinstance(data, list): return [_recursive_convert(item) for item in data] return data # 替换原始响应内容为转换后的JSON response._content = str(_recursive_convert(response.json())).encode('utf-8') # 给requests会话添加钩子,自动处理服务A的蛇形键 session = requests.Session() session.hooks['response'].append(convert_snake_response) # 请求服务A,直接得到驼峰格式数据 response_a = session.get('https://service-a.example.com/data') unified_a_data = response_a.json() # 请求服务B,直接使用原始驼峰数据 response_b = requests.get('https://service-b.example.com/data') unified_b_data = response_b.json() # 统一格式后直接整合排序 combined_data = [*unified_a_data, *unified_b_data] sorted_data = sorted(combined_data, key=lambda x: x['userName'])
三、整合时按需转换:生成器流式处理
如果需要处理大量数据,可用生成器按需转换,避免提前生成全量转换后的列表,降低内存占用:
def convert_snake_to_camel(item: Dict[str, Any]) -> Dict[str, Any]: return {snake_to_camel(k): v for k, v in item.items()} # 假设从服务A/B获取的是数据生成器 service_a_generator = (item for item in service_a_response.json()) service_b_generator = (item for item in service_b_response.json()) # 统一转换为驼峰的流式生成器 unified_generator = ( convert_snake_to_camel(item) if isinstance(item, dict) else item for item in service_a_generator ) # 合并服务B的驼峰数据 unified_generator = (item for item in unified_generator) unified_generator = (item for item in service_b_generator) # 直接基于生成器完成过滤和排序(仅迭代时才转换) filtered_sorted = sorted( (item for item in unified_generator if item['userId'] > 100), key=lambda x: x['userName'] )
内容的提问来源于stack exchange,提问作者Eduardo Elias Saléh
相关产品推荐
相关产品推荐

