Python json模块深度解析:从基础到高级,构建健壮的数据交换系统

import json# 基本数据类型转换data = {"name": "Alice","age": 30,"is_student": False,"courses": ["Math", "Physics", "Computer Science"],"address": None,"score": 95.5}json_str = json.dumps(data)print("基本序列化:")print(json_str)# 输出: {"name": "Alice", "age": 30, "is_student": false, "courses": ["Math", "Physics", "Computer Science"], "address": null, "score": 95.5}# 格式化输出（调试必备）pretty_json = json.dumps(data, indent=4, sort_keys=True)print("\n格式化输出:")print(pretty_json)# 输出:# {#     "address": null,#     "age": 30,#     "courses": [#         "Math",#         "Physics",#         "Computer Science"#     ],#     "is_student": false,#     "name": "Alice",#     "score": 95.5# }# 中文处理chinese_data = {"name": "张三", "city": "北京"}default_output = json.dumps(chinese_data)print("\n默认中文处理（转义）:")print(default_output)  # {"name": "\u5f20\u4e09", "city": "\u5317\u4eac"}correct_output = json.dumps(chinese_data, ensure_ascii=False)print("\n正确中文处理:")print(correct_output)  # {"name": "张三", "city": "北京"}

import json# 字符串解析json_str = '{"name": "Bob", "age": 25, "active": true, "tags": ["python", "web"]}'data = json.loads(json_str)print("解析结果:")print(f"类型: {type(data)}")  # <class 'dict'>print(f"内容: {data}")print(f"姓名: {data['name']}")  # Bobprint(f"标签: {data['tags'][0]}")  # python# 错误处理invalid_json = '{"name": "Charlie", age: 30}'# 键缺少引号try:    result = json.loads(invalid_json)except json.JSONDecodeError as e:    print(f"\nJSON解析错误:")    print(f"错误信息: {e.msg}")    print(f"错误位置: 第{e.lineno}行, 第{e.colno}列")    print(f"错误片段: {e.doc[e.pos:e.pos+20]}")

import jsonimport os# 准备测试数据data = {"project": "JSON Tutorial","version": "1.0","author": "Python与AI智能研习社","modules": ["json", "csv", "yaml"],"config": {"debug": True,"port": 8080,"host": "localhost"    }}# 写入JSON文件with open('config.json', 'w', encoding='utf-8') as f:    json.dump(data, f, ensure_ascii=False, indent=2)print("文件已写入: config.json")print(f"文件大小: {os.path.getsize('config.json')} 字节")# 读取JSON文件with open('config.json', 'r', encoding='utf-8') as f:    loaded_data = json.load(f)print("\n读取验证:")print(f"项目名称: {loaded_data['project']}")print(f"作者: {loaded_data['author']}")print(f"模块数量: {len(loaded_data['modules'])}")# 清理测试文件os.remove('config.json')

import jsonfrom datetime import datetimefrom decimal import DecimalclassUser:"""自定义用户类"""def__init__(self, name, email, created_at, balance):        self.name = name        self.email = email        self.created_at = created_at  # datetime对象        self.balance = balance  # Decimal对象def__repr__(self):returnf"User(name={self.name!r}, email={self.email!r})"defcustom_encoder(obj):"""自定义编码器函数"""if isinstance(obj, datetime):# 将datetime转换为ISO格式字符串return obj.isoformat()elif isinstance(obj, Decimal):# 将Decimal转换为字符串，避免精度丢失return str(obj)elif isinstance(obj, User):# 将User实例转换为字典return {"name": obj.name,"email": obj.email,"created_at": obj.created_at,"balance": obj.balance,"_type": "User"# 添加类型标识，便于反序列化        }# 无法处理的类型，让基类处理raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")# 创建测试对象user = User(    name="李四",    email="lisi@example.com",    created_at=datetime(2024, 3, 17, 14, 30, 0),    balance=Decimal("1234.56"))# 序列化自定义对象user_json = json.dumps(user, default=custom_encoder, ensure_ascii=False, indent=2)print("自定义对象序列化结果:")print(user_json)# 输出:# {#   "name": "李四",#   "email": "lisi@example.com",#   "created_at": "2024-03-17T14:30:00",#   "balance": "1234.56",#   "_type": "User"# }

import jsonfrom datetime import datetimefrom decimal import DecimalclassCustomJSONEncoder(json.JSONEncoder):"""自定义JSON编码器"""defdefault(self, obj):"""重写default方法"""if isinstance(obj, datetime):return {"_type": "datetime","value": obj.isoformat()            }elif isinstance(obj, Decimal):return {"_type": "decimal","value": str(obj)            }elif hasattr(obj, '__dict__'):# 处理任意自定义对象（有__dict__属性）            result = obj.__dict__.copy()            result['_type'] = type(obj).__name__return result# 调用父类方法处理其他类型return super().default(obj)# 使用自定义编码器encoder = CustomJSONEncoder(indent=2, ensure_ascii=False)encoded = encoder.encode(user)print("\n使用CustomJSONEncoder:")print(encoded)

defcustom_decoder(dct):"""自定义解码器函数"""# 检查类型标识if'_type'in dct:        type_name = dct['_type']if type_name == 'User':# 还原User对象return User(                name=dct['name'],                email=dct['email'],                created_at=datetime.fromisoformat(dct['created_at']),                balance=Decimal(dct['balance'])            )elif type_name == 'datetime':# 还原datetime对象return datetime.fromisoformat(dct['value'])elif type_name == 'decimal':# 还原Decimal对象return Decimal(dct['value'])# 没有类型标识，返回原字典return dct# 测试反序列化decoded_user = json.loads(user_json, object_hook=custom_decoder)print("\n反序列化结果验证:")print(f"类型: {type(decoded_user)}")  # <class '__main__.User'>print(f"姓名: {decoded_user.name}")  # 李四print(f"余额类型: {type(decoded_user.balance)}")  # <class 'decimal.Decimal'>print(f"创建时间: {decoded_user.created_at}")

from collections import OrderedDict# 保留原始键值对顺序ordered_json = '{"z": 1, "a": 2, "c": 3}'# 使用object_pairs_hook保留顺序data = json.loads(ordered_json, object_pairs_hook=OrderedDict)print("保留顺序的解析:")for key, value in data.items():    print(f"{key}: {value}")# 输出顺序: z, a, c （与JSON中顺序一致）# 对比默认解析（Python 3.7+字典保持插入顺序，但不保证与JSON一致）default_data = json.loads(ordered_json)print("\n默认解析（Python 3.7+保持插入顺序）:")for key, value in default_data.items():    print(f"{key}: {value}")

import jsonimport ijson  # 需要安装: pip install ijsondefprocess_large_json_file(file_path):"""流式处理大型JSON文件"""with open(file_path, 'r', encoding='utf-8') as f:# 使用ijson进行流式解析        objects = ijson.items(f, 'item')  # 假设JSON数组的键是'item'for obj in objects:# 逐个处理对象，避免一次性加载到内存            process_item(obj)    print("流式处理完成")defprocess_item(item):"""处理单个数据项（示例）"""# 这里是业务处理逻辑pass# 传统方法（内存危险）：# with open('large.json', 'r') as f:#     data = json.load(f)  # 可能内存溢出

defbatch_process_json(input_file, output_file, batch_size=1000):"""分批处理JSON数据"""with open(input_file, 'r', encoding='utf-8') as f_in:# 逐行读取（假设每行是一个JSON对象）        batch = []for line_num, line in enumerate(f_in, 1):try:                obj = json.loads(line.strip())                batch.append(obj)# 达到批次大小时处理if len(batch) >= batch_size:                    process_batch(batch, line_num - batch_size, line_num)                    batch.clear()except json.JSONDecodeError as e:                print(f"第{line_num}行解析错误: {e}")# 处理剩余数据if batch:            process_batch(batch, line_num - len(batch) + 1, line_num)    print("分批处理完成")defprocess_batch(batch, start_line, end_line):"""处理一个批次的数据"""    print(f"处理批次: 第{start_line}-{end_line}行，共{len(batch)}条记录")# 这里实现具体的业务逻辑

# 安装: pip install orjsonimport orjsonimport timedata = {"name": "性能测试","values": [i for i in range(100000)],"nested": {"level1": {"level2": {"level3": "deep"}}}}# 标准库性能start = time.time()json_str = json.dumps(data)json_time = time.time() - start# orjson性能start = time.time()orjson_str = orjson.dumps(data).decode('utf-8')orjson_time = time.time() - startprint(f"标准库序列化时间: {json_time:.4f}秒")print(f"orjson序列化时间: {orjson_time:.4f}秒")print(f"性能提升: {(json_time/orjson_time):.1f}倍")# orjson特性：# 1. 默认输出bytes而不是str# 2. 支持datetime、UUID、numpy数组等# 3. 不支持自定义编码器（设计取舍）

# 安装: pip install ujsonimport ujson# 使用方式与标准库基本相同fast_json = ujson.dumps(data)fast_data = ujson.loads(fast_json)# ujson优势：# 1. 比标准库快2-3倍# 2. API与标准库高度兼容# 3. 支持大整数不溢出

defgenerate_large_json_stream(items):"""生成JSON数据流（不一次性构建完整JSON）"""yield'['for i, item in enumerate(items):if i > 0:yield','yield json.dumps(item)yield']'# 使用示例items = [{"id": i, "data": "value" + str(i)} for i in range(1000000)]with open('large_output.json', 'w', encoding='utf-8') as f:for chunk in generate_large_json_stream(items):        f.write(chunk)    print("流式写入完成，内存占用恒定")

defselective_serialize(obj, include_fields=None, exclude_fields=None):"""选择性序列化，只包含需要的字段"""if include_fields isnotNone:# 只包含指定字段        filtered = {k: v for k, v in obj.items() if k in include_fields}return json.dumps(filtered)elif exclude_fields isnotNone:# 排除指定字段        filtered = {k: v for k, v in obj.items() if k notin exclude_fields}return json.dumps(filtered)else:# 包含所有字段return json.dumps(obj)# 使用示例user_data = {"id": 123,"name": "王五","email": "wangwu@example.com","password_hash": "abc123","created_at": "2024-03-17","last_login": "2024-03-17 14:30:00"}# 只包含公开字段（排除敏感信息）public_json = selective_serialize(    user_data,     include_fields=["id", "name", "email", "created_at"])print("安全序列化结果:")print(public_json)# {"id": 123, "name": "王五", "email": "wangwu@example.com", "created_at": "2024-03-17"}

# 危险：直接拼接用户输入user_input = '{"malicious": true}'dangerous_json = '{"data": ' + user_input + '}'# 可能被注入# 安全解析示例defsafe_json_parse(input_str, default=None):"""安全解析JSON，防止注入"""try:return json.loads(input_str)except (json.JSONDecodeError, TypeError):return default# 使用示例user_provided = '{"name": "test", "extra": "malicious"}}'# 多了一个大括号result = safe_json_parse(user_provided, default={"error": "invalid"})print(f"安全解析结果: {result}")

import jsonimport demjson3  # 需要安装: pip install demjson3defvalidate_and_parse_strict(json_str):"""严格验证和解析JSON"""try:# 使用demjson3进行严格验证return demjson3.decode(json_str, strict=True)except demjson3.JSONDecodeError as e:        print(f"严格模式验证失败: {e}")# 记录日志，返回安全默认值returnNone# 测试严格模式test_cases = ['{"valid": true}',  # 有效'{"invalid": 0123}',  # 无效：数字有前导零'{"duplicate": 1, "duplicate": 2}',  # 无效：重复键'{"trailing": "comma",}',  # 无效：尾部逗号]for i, test in enumerate(test_cases, 1):    result = validate_and_parse_strict(test)    print(f"测试{i}: {test[:30]}... -> {'有效'if result else'无效'}")

classSafeJSONDecoder(json.JSONDecoder):"""安全的JSON解码器，防止资源耗尽"""def__init__(self, *args, max_depth=100, max_str_len=10**7, **kwargs):        self.max_depth = max_depth        self.max_str_len = max_str_len        super().__init__(*args, **kwargs)defdecode(self, s, **kwargs):# 检查字符串长度if len(s) > self.max_str_len:raise ValueError(f"JSON字符串过长（{len(s)} > {self.max_str_len}）")# 调用父类解析，并在解析过程中检查深度return super().decode(s, **kwargs)# 使用安全解码器safe_decoder = SafeJSONDecoder(max_depth=50, max_str_len=10**6)try:# 尝试解析可能深度嵌套的JSON    deep_json = '{' * 100 + '"key": "value"' + '}' * 100    result = json.loads(deep_json, cls=safe_decoder)except (ValueError, RecursionError) as e:    print(f"安全检查拦截: {e}")

import jsonimport requestsfrom typing import Dict, AnyclassAPIClient:"""RESTful API客户端"""def__init__(self, base_url: str, timeout: int = 30):        self.base_url = base_url        self.timeout = timeout        self.session = requests.Session()defget(self, endpoint: str, params: Dict = None) -> Dict[str, Any]:"""发送GET请求"""        url = f"{self.base_url}/{endpoint}"try:            response = self.session.get(                url,                 params=params,                 timeout=self.timeout            )            response.raise_for_status()# 自动解析JSON响应return response.json()except requests.exceptions.RequestException as e:            print(f"API请求失败: {e}")return {"error": str(e), "success": False}defpost(self, endpoint: str, data: Dict) -> Dict[str, Any]:"""发送POST请求（JSON格式）"""        url = f"{self.base_url}/{endpoint}"try:            response = self.session.post(                url,                json=data,  # requests自动序列化为JSON                timeout=self.timeout            )            response.raise_for_status()return response.json()except requests.exceptions.RequestException as e:            print(f"API请求失败: {e}")return {"error": str(e), "success": False}defpost_with_custom_serialization(self, endpoint: str, obj) -> Dict[str, Any]:"""发送POST请求（自定义序列化）"""        url = f"{self.base_url}/{endpoint}"# 自定义序列化        json_str = json.dumps(obj, default=custom_encoder, ensure_ascii=False)try:            response = self.session.post(                url,                data=json_str,                headers={'Content-Type': 'application/json'},                timeout=self.timeout            )            response.raise_for_status()return response.json()except requests.exceptions.RequestException as e:            print(f"API请求失败: {e}")return {"error": str(e), "success": False}# 使用示例# client = APIClient("https://api.example.com")# user_data = client.get("users/123")# new_user = client.post("users", {"name": "新用户", "email": "new@example.com"})

import jsonimport osfrom typing import Dict, Anyfrom pathlib import PathclassConfigManager:"""JSON配置文件管理器"""def__init__(self, config_path: str = "config.json"):        self.config_path = Path(config_path)        self.config = self.load_config()defload_config(self, default_config: Dict = None) -> Dict[str, Any]:"""加载配置文件"""ifnot self.config_path.exists():            print(f"配置文件不存在: {self.config_path}")if default_config:                self.save_config(default_config)return default_configelse:return {}try:with open(self.config_path, 'r', encoding='utf-8') as f:                config = json.load(f)                print(f"配置文件加载成功: {self.config_path}")return configexcept (json.JSONDecodeError, OSError) as e:            print(f"配置文件解析失败: {e}")if default_config:                print("使用默认配置")return default_configelse:return {}defsave_config(self, config: Dict = None) -> bool:"""保存配置文件"""if config isNone:            config = self.configtry:# 创建目录（如果需要）            self.config_path.parent.mkdir(parents=True, exist_ok=True)with open(self.config_path, 'w', encoding='utf-8') as f:                json.dump(config, f, ensure_ascii=False, indent=2)            print(f"配置文件保存成功: {self.config_path}")returnTrueexcept OSError as e:            print(f"配置文件保存失败: {e}")returnFalsedefget(self, key: str, default=None) -> Any:"""获取配置值"""return self.config.get(key, default)defset(self, key: str, value: Any, save: bool = True) -> None:"""设置配置值"""        self.config[key] = valueif save:            self.save_config()defupdate(self, updates: Dict, save: bool = True) -> None:"""批量更新配置"""        self.config.update(updates)if save:            self.save_config()# 使用示例default_config = {"app": {"name": "MyApp","version": "1.0.0","debug": True    },"database": {"host": "localhost","port": 5432,"name": "mydb"    },"server": {"host": "0.0.0.0","port": 8080    }}config_manager = ConfigManager("myapp/config.json")app_name = config_manager.get("app.name", "默认应用")print(f"应用名称: {app_name}")# 更新配置config_manager.set("app.debug", False)config_manager.update({"server": {"port": 9000}})

import jsonimport loggingimport sysfrom datetime import datetimefrom typing import Dict, AnyclassJSONFormatter(logging.Formatter):"""JSON格式的日志格式化器"""defformat(self, record: logging.LogRecord) -> str:"""格式化日志记录为JSON"""        log_data = {"timestamp": datetime.utcnow().isoformat() + "Z","level": record.levelname,"logger": record.name,"message": record.getMessage(),"module": record.module,"function": record.funcName,"line": record.lineno        }# 添加额外字段if hasattr(record, 'extra'):            log_data.update(record.extra)# 处理异常if record.exc_info:            log_data['exception'] = self.formatException(record.exc_info)return json.dumps(log_data, ensure_ascii=False)defformatException(self, exc_info) -> Dict[str, Any]:"""格式化异常信息"""        ex_type, ex_value, ex_traceback = exc_inforeturn {"type": ex_type.__name__,"message": str(ex_value),"traceback": self.format_traceback(ex_traceback)        }defformat_traceback(self, traceback) -> list:"""格式化跟踪栈"""# 这里简化处理，实际可以提取更多信息import traceback as tbreturn tb.format_tb(traceback)# 配置日志系统defsetup_logging():"""设置JSON格式的日志"""    logger = logging.getLogger()    logger.setLevel(logging.INFO)# 移除现有处理器    logger.handlers.clear()# 控制台输出（JSON格式）    console_handler = logging.StreamHandler(sys.stdout)    console_handler.setFormatter(JSONFormatter())    logger.addHandler(console_handler)# 文件输出（JSON Lines格式）    file_handler = logging.FileHandler("app.log", encoding='utf-8')    file_handler.setFormatter(JSONFormatter())    logger.addHandler(file_handler)return logger# 使用示例logger = setup_logging()# 普通日志logger.info("应用启动成功")# 带上下文的日志logger.info("用户登录", extra={"user_id": 12345,"username": "zhangsan","ip_address": "192.168.1.100"})# 错误日志try:    result = 1 / 0except ZeroDivisionError:    logger.error("除零错误", exc_info=True)

import jsonfrom datetime import datetimefrom dateutil import parser  # 需要安装: pip install python-dateutil# 序列化defserialize_datetime(obj):if isinstance(obj, datetime):return obj.isoformat()raise TypeError(f"Type {type(obj)} not serializable")# 反序列化defdeserialize_datetime(dct):for key, value in dct.items():# 尝试解析ISO格式的日期时间字符串if isinstance(value, str):try:# 尝试解析多种格式                dct[key] = parser.isoparse(value)except (ValueError, TypeError):# 不是日期时间字符串，保持原样passreturn dct# 使用示例data = {"event": "meeting", "time": datetime.now()}json_str = json.dumps(data, default=serialize_datetime)print(f"序列化: {json_str}")parsed = json.loads(json_str, object_hook=deserialize_datetime)print(f"反序列化时间类型: {type(parsed['time'])}")

defgenerate_large_data_stream(count):"""生成大型数据流"""for i in range(count):yield {"id": i,"data": f"item_{i}","timestamp": datetime.now().isoformat()        }defsave_json_stream(stream, file_path):"""流式保存JSON数据"""with open(file_path, 'w', encoding='utf-8') as f:        f.write('[\n')        first = Truefor item in stream:ifnot first:                f.write(',\n')            json_line = json.dumps(item, ensure_ascii=False)            f.write('  ' + json_line)            first = False        f.write('\n]')    print(f"流式保存完成: {file_path}")# 使用示例（生成100万条记录）# data_stream = generate_large_data_stream(1000000)# save_json_stream(data_stream, "large_data.json")

import jsonimport jsonschema  # 需要安装: pip install jsonschema# 定义Schemauser_schema = {"type": "object","properties": {"name": {"type": "string", "minLength": 1},"age": {"type": "integer", "minimum": 0, "maximum": 150},"email": {"type": "string", "format": "email"},"is_active": {"type": "boolean"},"tags": {"type": "array","items": {"type": "string"},"uniqueItems": True        }    },"required": ["name", "email"],"additionalProperties": False}defvalidate_user_data(data):"""验证用户数据是否符合Schema"""try:        jsonschema.validate(instance=data, schema=user_schema)returnTrue, "验证通过"except jsonschema.ValidationError as e:returnFalse, f"验证失败: {e.message}"except json.JSONDecodeError as e:returnFalse, f"JSON解析失败: {e}"# 测试验证test_cases = [    {"name": "张三", "email": "zhangsan@example.com", "age": 25},  # 有效    {"name": "", "email": "invalid"},  # 无效：姓名为空，邮箱格式错误    {"name": "李四", "age": "二十五"},  # 无效：年龄不是整数]for i, test_data in enumerate(test_cases, 1):    is_valid, message = validate_user_data(test_data)    print(f"测试{i}: {test_data.get('name', '无名')} -> {message}")