Python json模块完整教程

Python json模块完整教程

项目简介

本项目是一个完整的Python json模块教程，涵盖了从基础到高级的所有知识点。

所有代码都基于生产环境的实际场景，配有详细的逐行注释。

环境要求

• Python 3.12
• 虚拟环境：.venv

安装步骤

教程内容

第1章：JSON基础和json模块概述

1.1 JSON简介

JSON (JavaScript Object Notation) 是一种轻量级的数据交换格式，易于人阅读和编写，同时也易于机器解析和生成。

JSON的特点：

• 轻量级：文本格式，体积小
• 语言无关：几乎所有编程语言都支持
• 易于解析：结构简单，解析速度快
• 自描述性：数据包含结构信息

# JSON (JavaScript Object Notation) 是一种轻量级的数据交换格式
# 易于人阅读和编写，同时也易于机器解析和生成
# 基于JavaScript Programming Language, Standard ECMA-262 3rd Edition

# JSON的特点：
# – 轻量级：文本格式，体积小
# – 语言无关：几乎所有编程语言都支持
# – 易于解析：结构简单，解析速度快
# – 自描述性：数据包含结构信息

print("JSON (JavaScript Object Notation) 是一种轻量级的数据交换格式")
print("特点：轻量级、语言无关、易于解析、自描述性")

1.2 JSON的数据类型

JSON支持6种数据类型：

# JSON支持6种数据类型：
# 1. 字符串 (string) – 使用双引号
# 2. 数字 (number) – 整数或浮点数
# 3. 布尔值 (boolean) – true 或 false
# 4. 数组 (array) – 有序集合，用 [] 表示
# 5. 对象 (object) – 键值对，用 {} 表示
# 6. null – 空值

# 示例：JSON数据结构
json_example = '''
{
"name": "张三",
"age": 28,
"is_active": true,
"scores": [95, 88, 92],
"address": {
"city": "北京",
"zip": "100000"
},
"hobbies": null
}
'''

print("JSON支持的数据类型：")
print("- 字符串 (string) – 使用双引号")
print("- 数字 (number) – 整数或浮点数")
print("- 布尔值 (boolean) – true 或 false")
print("- 数组 (array) – 有序集合，用 []")
print("- 对象 (object) – 键值对，用 {}")
print("- null – 空值")

1.3 Python json模块核心函数

json模块提供4个核心函数：

# json模块提供4个核心函数：
# 1. json.loads() – 将JSON字符串解析为Python对象
# 2. json.load() – 从文件对象读取JSON并解析为Python对象
# 3. json.dumps() – 将Python对象序列化为JSON字符串
# 4. json.dump() – 将Python对象序列化为JSON并写入文件

# 函数对比图：
# 解析 (Parsing) – JSON -> Python
# loads() : JSON字符串 -> Python对象
# load() : JSON文件 -> Python对象

# 序列化 (Serialization) – Python -> JSON
# dumps() : Python对象 -> JSON字符串
# dump() : Python对象 -> JSON文件

print("json模块的4个核心函数：")
print(" loads() – JSON字符串 -> Python对象 (解析)")
print(" load() – JSON文件 -> Python对象 (解析)")
print(" dumps() – Python对象 -> JSON字符串 (序列化)")
print(" dump() – Python对象 -> JSON文件 (序列化)")

1.4 JSON与Python数据类型映射

# JSON类型 -> Python类型
# object -> dict
# array -> list
# string -> str
# number (int) -> int
# number (real) -> float
# true/false -> True/False
# null -> None

# 示例：类型映射
import json

# JSON对象 -> Python字典
json_obj = '{"name": "李四", "age": 30}'
py_dict = json.loads(json_obj)
print(f"JSON对象 -> Python字典: {type(py_dict).__name__}")

# JSON数组 -> Python列表
json_arr = '[1, 2, 3, 4, 5]'
py_list = json.loads(json_arr)
print(f"JSON数组 -> Python列表: {type(py_list).__name__}")

# JSON字符串 -> Python字符串
json_str = '"Hello, Python"'
py_str = json.loads(json_str)
print(f"JSON字符串 -> Python字符串: {type(py_str).__name__}")

# JSON数字 -> Python数字
json_num = '42'
py_num = json.loads(json_num)
print(f"JSON数字 -> Python数字: {type(py_num).__name__}")

# JSON布尔值 -> Python布尔值
json_bool = 'true'
py_bool = json.loads(json_bool)
print(f"JSON布尔值 -> Python布尔值: {type(py_bool).__name__}")

# JSON null -> Python None
json_null = 'null'
py_none = json.loads(json_null)
print(f"JSON null -> Python None: {type(py_none).__name__}")

1.5 实际应用场景

场景1: Web API数据交换

• 前端JavaScript和后端Python通过JSON交换数据

场景2: 配置文件存储

• 存储应用配置信息

场景3: 日志数据格式

• 结构化的日志记录

# 场景1: Web API数据交换
# 前端JavaScript和后端Python通过JSON交换数据

api_response = '''
{
"status": "success",
"data": {
"user_id": 1001,
"username": "user001",
"permissions": ["read", "write", "delete"]
},
"timestamp": "2024-01-15T10:30:00Z"
}
'''

# 场景2: 配置文件存储
config = {
"database": {
"host": "localhost",
"port": 3306,
"username": "admin",
"password": "secret123"
},
"cache": {
"enabled": True,
"ttl": 3600
}
}

# 场景3: 日志数据格式
log_entry = {
"level": "ERROR",
"timestamp": "2024-01-15 14:30:45",
"module": "user_service",
"message": "用户登录失败",
"details": {
"user_id": 1001,
"ip": "192.168.1.100"
}
}

print("常见应用场景：")
print(" 1. Web API数据交换 (前后端通信)")
print(" 2. 配置文件存储")
print(" 3. 日志数据格式")
print(" 4. 数据库数据导出/导入")
print(" 5. 微服务间通信")
print(" 6. 文件数据持久化")

1.6 json模块基础使用示例

# 导入json模块
import json

# 示例1: 使用json.loads() 解析JSON字符串
print("\\n示例1: json.loads() – 解析JSON字符串")
user_json = '{"id": 1, "name": "王五", "email": "wangwu@example.com"}'
print(f"JSON字符串: {user_json}")

# 将JSON字符串解析为Python字典
user_dict = json.loads(user_json)
print(f"解析结果类型: {type(user_dict)}")
print(f"解析结果内容: {user_dict}")
print(f"访问字典值: name = {user_dict['name']}")

# 示例2: 使用json.dumps() 序列化为JSON字符串
print("\\n示例2: json.dumps() – 序列化为JSON字符串")
product = {
"id": 101,
"name": "笔记本电脑",
"price": 5999.99,
"in_stock": True,
"tags": ["电子", "办公", "便携"]
}
print(f"Python字典: {product}")

# 将Python字典序列化为JSON字符串
product_json = json.dumps(product)
print(f"序列化结果: {product_json}")
print(f"序列化结果类型: {type(product_json)}")

# 示例3: 使用json.load() 从文件读取JSON
print("\\n示例3: json.load() – 从文件读取JSON")
import tempfile
import os

# 创建临时JSON文件
temp_dir = tempfile.gettempdir()
temp_file = os.path.join(temp_dir, "test_data.json")

test_data = {
"students": [
{"name": "学生1", "score": 85},
{"name": "学生2", "score": 92}
]
}

# 写入临时文件
with open(temp_file, 'w', encoding='utf-8') as f:
json.dump(test_data, f, ensure_ascii=False, indent=2)

# 从文件读取JSON
with open(temp_file, 'r', encoding='utf-8') as f:
loaded_data = json.load(f)
print(f"从文件读取的数据: {loaded_data}")

# 清理临时文件
os.remove(temp_file)

# 示例4: 使用json.dump() 写入JSON到文件
print("\\n示例4: json.dump() – 写入JSON到文件")
order = {
"order_id": "ORD20240115001",
"customer": "赵六",
"items": [
{"product": "鼠标", "quantity": 2, "price": 89.00},
{"product": "键盘", "quantity": 1, "price": 299.00}
],
"total_amount": 477.00,
"status": "pending"
}

# 将订单数据写入文件
output_file = os.path.join(temp_dir, "order.json")
with open(output_file, 'w', encoding='utf-8') as f:
json.dump(order, f, ensure_ascii=False, indent=4)
print(f"订单数据已保存到: {output_file}")

# 读取并验证
with open(output_file, 'r', encoding='utf-8') as f:
saved_order = json.load(f)
print(f"保存的订单: {saved_order['order_id']}")

# 清理
os.remove(output_file)

本章小结

第2章：json.loads() – JSON解析

2.1 json.loads() 函数语法

json.loads(s, *, cls=None, object_hook=None, parse_float=None,
parse_int=None, parse_constant=None, strict=True,
object_pairs_hook=None)

参数说明：

• s (必需): JSON字符串
• cls: 自定义JSONDecoder类
• object_hook: 解析对象时的回调函数
• parse_float: 自定义浮点数解析
• parse_int: 自定义整数解析
• parse_constant: 自定义常量解析
• strict: 是否允许控制字符
• object_pairs_hook: 处理有序键值对的函数

2.2 基础解析示例

import json

# 解析JSON对象
user_json = '{"id": 1, "name": "张三", "email": "zhangsan@example.com"}'
user = json.loads(user_json)
print(f"用户ID: {user['id']}")
print(f"用户名: {user['name']}")

# 解析JSON数组
scores_json = '[95, 88, 92, 78, 85]'
scores = json.loads(scores_json)
print(f"最高分: {max(scores)}")
print(f"平均分: {sum(scores) / len(scores)}")

# 解析嵌套JSON
order_json = '''
{
"order_id": "ORD001",
"customer": "李四",
"items": [
{"product": "手机", "quantity": 1, "price": 5999.00},
{"product": "耳机", "quantity": 2, "price": 299.00}
],
"total_amount": 6597.00
}
'''
order = json.loads(order_json)
print(f"订单ID: {order['order_id']}")
print(f"商品数量: {len(order['items'])}")
print(f"订单总额: {order['total_amount']}")

2.3 object_hook参数

import json
from datetime import datetime

# 使用object_hook自定义对象转换
def custom_object_hook(obj):
"""自定义对象转换函数"""
if 'created_at' in obj:
obj['created_at'] = datetime.fromisoformat(obj['created_at'])
return obj

data_json = '''
{
"user_id": 1001,
"username": "wangwu",
"created_at": "2024-01-15T10:30:00"
}
'''

data = json.loads(data_json, object_hook=custom_object_hook)
print(f"创建时间类型: {type(data['created_at'])}")
print(f"创建时间: {data['created_at']}")

2.4 parse_float和parse_int参数

import json
from decimal import Decimal

# 使用parse_float自定义浮点数解析
def decimal_parse_float(num_str):
"""使用Decimal解析浮点数"""
return Decimal(num_str)

price_json = '{"product": "手机", "price": 5999.99, "discount": 0.95}'
price_data = json.loads(price_json, parse_float=Decimal)
print(f"价格类型: {type(price_data['price'])}")
print(f"价格: {price_data['price']}")

# 使用parse_int自定义整数解析
def custom_parse_int(num_str):
"""自定义整数解析"""
return int(num_str) * 2 # 将整数乘以2

count_json = '{"count": "100", "total": 200}'
count_data = json.loads(count_json, parse_int=custom_parse_int)
print(f"计数: {count_data['count']}")

2.5 parse_constant参数

import json
import math

# 使用parse_constant处理特殊常量
def custom_parse_constant(const_str):
"""自定义常量解析"""
if const_str == 'NaN':
return float('nan')
elif const_str == 'Infinity':
return float('inf')
elif const_str == '-Infinity':
return float('-inf')
raise ValueError(f"未知常量: {const_str}")

special_json = '{"value1": NaN, "value2": Infinity, "value3": -Infinity}'
special_data = json.loads(special_json, parse_constant=custom_parse_constant)
print(f"NaN值: {math.isnan(special_data['value1'])}")
print(f"无穷大: {math.isinf(special_data['value2'])}")

2.6 strict参数

import json

# strict=True (默认): 不允许控制字符
# strict=False: 允许控制字符

# 包含控制字符的JSON (需要strict=False)
json_with_control = '{"message": "Hello\\\\nWorld\\\\t!"}'

# 这会抛出异常
try:
data = json.loads(json_with_control, strict=True)
except json.JSONDecodeError as e:
print(f"严格模式错误: {e.msg}")

# 使用strict=False
data = json.loads(json_with_control, strict=False)
print(f"消息内容: {repr(data['message'])}")

2.7 object_pairs_hook参数

import json
from collections import OrderedDict

# 使用object_pairs_hook保持键的顺序
def ordered_pairs_hook(pairs):
"""将键值对转换为有序字典"""
return OrderedDict(pairs)

ordered_json = '{"a": 1, "b": 2, "c": 3, "d": 4}'
ordered_data = json.loads(ordered_json, object_pairs_hook=ordered_pairs_hook)
print(f"数据类型: {type(ordered_data)}")
print(f"键顺序: {list(ordered_data.keys())}")

2.8 实际生产场景应用

import json
from datetime import datetime
from decimal import Decimal

# 实际场景：API响应解析
def parse_api_response(response_json):
"""解析API响应"""
def custom_object_hook(obj):
# 解析时间戳
if 'timestamp' in obj:
obj['timestamp'] = datetime.fromisoformat(obj['timestamp'])
# 解析金额
if 'amount' in obj:
obj['amount'] = Decimal(str(obj['amount']))
return obj

return json.loads(response_json, object_hook=custom_object_hook)

api_response = '''
{
"status": "success",
"data": {
"order_id": "ORD001",
"amount": 5999.99,
"timestamp": "2024-01-15T10:30:00"
}
}
'''

parsed = parse_api_response(api_response)
print(f"订单ID: {parsed['data']['order_id']}")
print(f"金额类型: {type(parsed['data']['amount'])}")
print(f"时间类型: {type(parsed['data']['timestamp'])}")

第3章：json.load() – 从文件读取JSON

3.1 json.load() 函数语法

json.load(fp, *, cls=None, object_hook=None, parse_float=None,
parse_int=None, parse_constant=None, strict=True,
object_pairs_hook=None)

参数说明：

• fp: 文件对象或类文件对象
• 其他参数与json.loads()相同

3.2 基础文件读取

import json
import tempfile
import os

# 创建示例JSON文件
temp_dir = tempfile.gettempdir()
test_file = os.path.join(temp_dir, "test.json")

data = {
"users": [
{"id": 1, "name": "张三", "email": "zhangsan@example.com"},
{"id": 2, "name": "李四", "email": "lisi@example.com"}
],
"total": 2
}

# 写入文件
with open(test_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)

# 从文件读取
with open(test_file, 'r', encoding='utf-8') as f:
loaded = json.load(f)
print(f"用户数量: {loaded['total']}")
for user in loaded['users']:
print(f" – {user['name']} ({user['email']})")

# 清理
os.remove(test_file)

3.3 从网络流读取

import json
from urllib.request import urlopen

# 示例：从API读取JSON数据
def fetch_json_data(url):
"""从URL获取JSON数据"""
with urlopen(url) as response:
data = json.load(response)
return data

# 注意：实际使用时需要有效的API URL
# api_url = "https://api.example.com/data"
# data = fetch_json_data(api_url)

3.4 大文件处理

import json

def process_large_json_file(filepath):
"""处理大JSON文件"""
with open(filepath, 'r', encoding='utf-8') as f:
# 逐行读取和处理
for line in f:
if line.strip():
try:
data = json.loads(line)
# 处理数据
process_item(data)
except json.JSONDecodeError:
continue

def process_item(item):
"""处理单个项目"""
print(f"处理项目: {item.get('id')}")

3.5 流式读取

import json

def stream_json_from_file(filepath):
"""流式读取JSON文件"""
with open(filepath, 'r', encoding='utf-8') as f:
decoder = json.JSONDecoder()
buffer = ""

for chunk in iter(lambda: f.read(1024), ""):
buffer += chunk
while buffer:
try:
obj, idx = decoder.raw_decode(buffer)
yield obj
buffer = buffer[idx:].lstrip()
except json.JSONDecodeError:
break

# 使用示例
# for obj in stream_json_from_file("large_file.json"):
# print(obj)

3.6 文件编码处理

import json

def read_json_with_encoding(filepath, encoding='utf-8'):
"""读取指定编码的JSON文件"""
try:
with open(filepath, 'r', encoding=encoding) as f:
return json.load(f)
except UnicodeDecodeError:
# 尝试其他编码
for enc in ['utf-8-sig', 'gbk', 'latin-1']:
try:
with open(filepath, 'r', encoding=enc) as f:
return json.load(f)
except UnicodeDecodeError:
continue
raise ValueError("无法确定文件编码")

3.7 实际生产场景应用

import json
import os
from datetime import datetime

class ConfigManager:
"""配置管理器"""

def __init__(self, config_file):
self.config_file = config_file
self.config = None

def load_config(self):
"""加载配置文件"""
if os.path.exists(self.config_file):
with open(self.config_file, 'r', encoding='utf-8') as f:
self.config = json.load(f)
else:
self.config = {}
return self.config

def get(self, key, default=None):
"""获取配置值"""
if self.config is None:
self.load_config()
return self.config.get(key, default)

def save_config(self):
"""保存配置文件"""
with open(self.config_file, 'w', encoding='utf-8') as f:
json.dump(self.config, f, ensure_ascii=False, indent=2)

# 使用示例
config_manager = ConfigManager("app_config.json")
config = config_manager.load_config()
db_host = config_manager.get('database.host', 'localhost')
print(f"数据库主机: {db_host}")

第4章：json.dumps() – 序列化为JSON字符串

4.1 json.dumps() 函数语法

json.dumps(obj, *, skipkeys=False, ensure_ascii=True,
check_circular=True, allow_nan=True, cls=None,
indent=None, separators=None, default=None,
sort_keys=False, **kw)

参数说明：

• obj: 要序列化的Python对象
• skipkeys: 跳过非字符串键
• ensure_ascii: 转义非ASCII字符
• check_circular: 检查循环引用
• allow_nan: 允许NaN、Infinity等
• indent: 缩进空格数
• separators: 自定义分隔符
• default: 处理不可序列化对象的函数
• sort_keys: 排序键

4.2 基础序列化

import json

# 序列化字典
user = {"id": 1, "name": "张三", "email": "zhangsan@example.com"}
user_json = json.dumps(user)
print(f"序列化结果: {user_json}")

# 序列化列表
numbers = [1, 2, 3, 4, 5]
numbers_json = json.dumps(numbers)
print(f"列表序列化: {numbers_json}")

# 序列化嵌套结构
order = {
"order_id": "ORD001",
"items": [
{"product": "手机", "price": 5999.00},
{"product": "耳机", "price": 299.00}
],
"total": 6597.00
}
order_json = json.dumps(order)
print(f"订单JSON: {order_json}")

4.3 ensure_ascii参数

import json

# ensure_ascii=True (默认): 转义中文
data = {"name": "张三", "city": "北京"}
json_ascii = json.dumps(data, ensure_ascii=True)
print(f"ASCII转义: {json_ascii}")

# ensure_ascii=False: 保留中文
json_unicode = json.dumps(data, ensure_ascii=False)
print(f"保留中文: {json_unicode}")

4.4 indent参数

import json

data = {"name": "张三", "age": 28, "email": "zhangsan@example.com"}

# 无缩进（紧凑格式）
compact = json.dumps(data)
print(f"紧凑格式: {compact}")

# 使用缩进（格式化）
formatted = json.dumps(data, indent=2)
print(f"格式化:\\n{formatted}")

# 使用4空格缩进
formatted_4 = json.dumps(data, indent=4)
print(f"4空格缩进:\\n{formatted_4}")

4.5 separators参数

import json

data = {"name": "张三", "age": 28, "email": "zhangsan@example.com"}

# 默认分隔符
default = json.dumps(data)
print(f"默认分隔符: {default}")

# 自定义分隔符 (item_separator, key_separator)
custom = json.dumps(data, separators=(',', ':'))
print(f"自定义分隔符: {custom}")

# 去除空格
no_space = json.dumps(data, separators=(',', ':'))
print(f"无空格: {no_space}")

4.6 sort_keys参数

import json

data = {"z_name": "张三", "a_age": 28, "m_email": "zhangsan@example.com"}

# 无排序
no_sort = json.dumps(data)
print(f"无排序: {no_sort}")

# 键排序
sorted_keys = json.dumps(data, sort_keys=True)
print(f"键排序: {sorted_keys}")

4.7 default参数

import json
from datetime import datetime

# 自定义default函数
def custom_default(obj):
"""处理不可序列化的对象"""
if isinstance(obj, datetime):
return obj.isoformat()
raise TypeError(f"无法序列化 {type(obj)}")

data = {
"name": "张三",
"created_at": datetime.now(),
"updated_at": datetime.now()
}

json_str = json.dumps(data, default=custom_default)
print(f"自定义序列化: {json_str}")

4.8 allow_nan参数

import json
import math

data = {
"valid": 42,
"nan_value": float('nan'),
"inf_value": float('inf')
}

# allow_nan=True (默认)
with_nan = json.dumps(data, allow_nan=True)
print(f"允许NaN: {with_nan}")

# allow_nan=False
try:
strict = json.dumps(data, allow_nan=False)
except ValueError as e:
print(f"严格模式错误: {e}")

4.9 实际生产场景应用

import json
from datetime import datetime
from decimal import Decimal

class APIResponse:
"""API响应类"""

def __init__(self, status, data, timestamp=None):
self.status = status
self.data = data
self.timestamp = timestamp or datetime.now()

def to_json(self):
"""序列化为JSON"""
def custom_default(obj):
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return float(obj)
raise TypeError(f"无法序列化 {type(obj)}")

return json.dumps({
"status": self.status,
"data": self.data,
"timestamp": self.timestamp
}, default=custom_default, ensure_ascii=False, indent=2)

# 使用示例
response = APIResponse(
status="success",
data={"order_id": "ORD001", "amount": Decimal("5999.99")}
)
print(response.to_json())

第5章：json.dump() – 序列化到文件

5.1 json.dump() 函数语法

json.dump(obj, fp, *, skipkeys=False, ensure_ascii=True,
check_circular=True, allow_nan=True, cls=None,
indent=None, separators=None, default=None,
sort_keys=False, **kw)

参数说明：

• obj: 要序列化的Python对象
• fp: 文件对象
• 其他参数与json.dumps()相同

5.2 基础文件写入

import json
import tempfile
import os

# 创建示例数据
data = {
"users": [
{"id": 1, "name": "张三", "email": "zhangsan@example.com"},
{"id": 2, "name": "李四", "email": "lisi@example.com"}
],
"total": 2,
"created_at": "2024-01-15"
}

# 写入JSON文件
temp_dir = tempfile.gettempdir()
output_file = os.path.join(temp_dir, "output.json")

with open(output_file, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
print(f"数据已写入: {output_file}")

# 验证写入
with open(output_file, 'r', encoding='utf-8') as f:
loaded = json.load(f)
print(f"写入验证: {loaded['total']} 个用户")

# 清理
os.remove(output_file)

5.3 与json.dumps()对比

import json

data = {"name": "张三", "age": 28}

# 方法1: 使用json.dumps() + 文件写入
json_str = json.dumps(data, indent=2)
with open("file1.json", 'w', encoding='utf-8') as f:
f.write(json_str)

# 方法2: 使用json.dump()
with open("file2.json", 'w', encoding='utf-8') as f:
json.dump(data, f, indent=2)

# json.dump()更简洁，性能更好

5.4 文件编码处理

import json

def write_json_with_encoding(filepath, data, encoding='utf-8'):
"""写入指定编码的JSON文件"""
with open(filepath, 'w', encoding=encoding) as f:
json.dump(data, f, ensure_ascii=False, indent=2)

# 写入UTF-8文件
write_json_with_encoding("utf8.json", {"name": "张三"})

# 写入GBK文件（Windows中文环境）
write_json_with_encoding("gbk.json", {"name": "张三"}, encoding='gbk')

5.5 实际生产场景应用

import json
import os
from datetime import datetime

class DataPersister:
"""数据持久化器"""

def __init__(self, base_dir):
self.base_dir = base_dir
os.makedirs(base_dir, exist_ok=True)

def save_orders(self, orders, filename="orders.json"):
"""保存订单数据"""
filepath = os.path.join(self.base_dir, filename)
with open(filepath, 'w', encoding='utf-8') as f:
json.dump({
"saved_at": datetime.now().isoformat(),
"orders": orders
}, f, ensure_ascii=False, indent=2)
return filepath

def load_orders(self, filename="orders.json"):
"""加载订单数据"""
filepath = os.path.join(self.base_dir, filename)
if os.path.exists(filepath):
with open(filepath, 'r', encoding='utf-8') as f:
return json.load(f)
return {"saved_at": None, "orders": []}

# 使用示例
persister = DataPersister("data")
orders = [
{"order_id": "ORD001", "amount": 5999.99},
{"order_id": "ORD002", "amount": 299.00}
]
filepath = persister.save_orders(orders)
print(f"订单已保存到: {filepath}")

第6章：自定义编码器和解码器

6.1 JSONEncoder和JSONDecoder类

import json

# 自定义Encoder
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)

# 自定义Decoder
class CustomDecoder(json.JSONDecoder):
def __init__(self, *args, **kwargs):
super().__init__(object_hook=self.object_hook, *args, **kwargs)

def object_hook(self, obj):
if 'date' in obj:
obj['date'] = datetime.fromisoformat(obj['date'])
return obj

6.2 处理datetime对象

import json
from datetime import datetime

class DateTimeEncoder(json.JSONEncoder):
"""datetime编码器"""

def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)

data = {
"name": "张三",
"created_at": datetime.now(),
"updated_at": datetime.now()
}

json_str = json.dumps(data, cls=DateTimeEncoder)
print(f"datetime序列化: {json_str}")

6.3 处理自定义类

import json

class User:
def __init__(self, user_id, name, email):
self.user_id = user_id
self.name = name
self.email = email

def to_dict(self):
return {
"user_id": self.user_id,
"name": self.name,
"email": self.email
}

class UserEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, User):
return obj.to_dict()
return super().default(obj)

user = User(1, "张三", "zhangsan@example.com")
json_str = json.dumps(user, cls=UserEncoder)
print(f"自定义类序列化: {json_str}")

6.4 处理Decimal对象

import json
from decimal import Decimal

class DecimalEncoder(json.JSONEncoder):
"""Decimal编码器"""

def default(self, obj):
if isinstance(obj, Decimal):
return float(obj)
return super().default(obj)

data = {
"product": "手机",
"price": Decimal("5999.99"),
"discount": Decimal("0.95")
}

json_str = json.dumps(data, cls=DecimalEncoder)
print(f"Decimal序列化: {json_str}")

6.5 处理UUID对象

import json
import uuid

class UUIDEncoder(json.JSONEncoder):
"""UUID编码器"""

def default(self, obj):
if isinstance(obj, uuid.UUID):
return str(obj)
return super().default(obj)

data = {
"order_id": uuid.uuid4(),
"user_id": uuid.uuid4()
}

json_str = json.dumps(data, cls=UUIDEncoder)
print(f"UUID序列化: {json_str}")

6.6 处理set和frozenset

import json

class SetEncoder(json.JSONEncoder):
"""set编码器"""

def default(self, obj):
if isinstance(obj, (set, frozenset)):
return list(obj)
return super().default(obj)

data = {
"tags": {"python", "json", "tutorial"},
"permissions": frozenset(["read", "write"])
}

json_str = json.dumps(data, cls=SetEncoder)
print(f"set序列化: {json_str}")

6.7 处理嵌套对象

import json
from datetime import datetime

class NestedEncoder(json.JSONEncoder):
"""嵌套对象编码器"""

def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
if hasattr(obj, 'to_dict'):
return obj.to_dict()
return super().default(obj)

class Address:
def __init__(self, city, street):
self.city = city
self.street = street

def to_dict(self):
return {"city": self.city, "street": self.street}

class User:
def __init__(self, user_id, name, address):
self.user_id = user_id
self.name = name
self.address = address

def to_dict(self):
return {
"user_id": self.user_id,
"name": self.name,
"address": self.address.to_dict()
}

address = Address("北京", "朝阳区")
user = User(1, "张三", address)

json_str = json.dumps(user, cls=NestedEncoder)
print(f"嵌套对象序列化: {json_str}")

6.8 实际生产场景应用

import json
from datetime import datetime
from decimal import Decimal
import uuid

class Product:
def __init__(self, product_id, name, price, stock):
self.product_id = product_id
self.name = name
self.price = price
self.stock = stock
self.created_at = datetime.now()

def to_dict(self):
return {
"product_id": self.product_id,
"name": self.name,
"price": float(self.price),
"stock": self.stock,
"created_at": self.created_at.isoformat()
}

class Order:
def __init__(self, order_id, user_id, items):
self.order_id = order_id
self.user_id = user_id
self.items = items
self.created_at = datetime.now()

def to_dict(self):
return {
"order_id": str(self.order_id),
"user_id": self.user_id,
"items": [item.to_dict() for item in self.items],
"created_at": self.created_at.isoformat()
}

class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, (datetime, Decimal)):
return str(obj)
if isinstance(obj, uuid.UUID):
return str(obj)
if hasattr(obj, 'to_dict'):
return obj.to_dict()
return super().default(obj)

# 使用示例
product = Product(1, "手机", Decimal("5999.99"), 100)
order = Order(uuid.uuid4(), 1001, [product])

json_str = json.dumps(order, cls=CustomEncoder, indent=2)
print(json_str)

第7章：JSON Schema验证

7.1 JSON Schema基础概念

JSON Schema用于定义JSON数据的结构和约束：

• type: 数据类型
• required: 必需字段
• properties: 属性定义
• minimum/maximum: 数值范围
• minLength/maxLength: 字符串长度
• enum: 枚举值
• items: 数组元素定义

7.2 基本验证

from jsonschema import validate, ValidationError

# 定义Schema
schema = {
"type": "object",
"required": ["name", "email"],
"properties": {
"name": {"type": "string", "minLength": 1},
"email": {"type": "string", "format": "email"},
"age": {"type": "integer", "minimum": 0, "maximum": 150}
}
}

# 有效数据
valid_data = {
"name": "张三",
"email": "zhangsan@example.com",
"age": 28
}

validate(instance=valid_data, schema=schema)
print("✓ 有效数据")

# 无效数据
invalid_data = {
"name": "", # 长度不足
"email": "invalid-email"
}

try:
validate(instance=invalid_data, schema=schema)
except ValidationError as e:
print(f"✗ 无效数据: {e.message}")

7.3 数值验证

from jsonschema import validate, ValidationError

schema = {
"type": "object",
"properties": {
"price": {
"type": "number",
"minimum": 0,
"maximum": 1000000
},
"quantity": {
"type": "integer",
"minimum": 1,
"maximum": 1000
}
}
}

# 有效数据
valid = {"price": 99.99, "quantity": 5}
validate(instance=valid, schema=schema)
print("✓ 数值验证通过")

# 无效数据
invalid = {"price": –10, "quantity": 0}
try:
validate(instance=invalid, schema=schema)
except ValidationError as e:
print(f"✗ 数值验证失败: {e.message}")

7.4 字符串验证

from jsonschema import validate, ValidationError

schema = {
"type": "object",
"properties": {
"username": {
"type": "string",
"minLength": 3,
"maxLength": 20,
"pattern": "^[a-zA-Z0-9_]+$"
},
"email": {
"type": "string",
"format": "email"
},
"phone": {
"type": "string",
"pattern": "^1[3-9]\\\\d{9}$"
}
}
}

# 有效数据
valid = {
"username": "zhang_san",
"email": "zhangsan@example.com",
"phone": "13800138000"
}
validate(instance=valid, schema=schema)
print("✓ 字符串验证通过")

7.5 数组验证

from jsonschema import validate, ValidationError

schema = {
"type": "object",
"properties": {
"tags": {
"type": "array",
"minItems": 1,
"maxItems": 10,
"items": {"type": "string", "minLength": 1}
},
"scores": {
"type": "array",
"items": {"type": "number", "minimum": 0, "maximum": 100}
}
}
}

# 有效数据
valid = {
"tags": ["python", "json"],
"scores": [95, 88, 92]
}
validate(instance=valid, schema=schema)
print("✓ 数组验证通过")

7.6 枚举验证

from jsonschema import validate, ValidationError

schema = {
"type": "object",
"properties": {
"status": {
"type": "string",
"enum": ["pending", "paid", "shipped", "delivered"]
},
"priority": {
"type": "string",
"enum": ["low", "medium", "high", "critical"]
}
}
}

# 有效数据
valid = {"status": "paid", "priority": "high"}
validate(instance=valid, schema=schema)
print("✓ 枚举验证通过")

# 无效数据
invalid = {"status": "unknown"}
try:
validate(instance=invalid, schema=schema)
except ValidationError as e:
print(f"✗ 枚举验证失败: {e.message}")

7.7 实际生产场景应用

from jsonschema import validate, ValidationError
import json

class DataValidator:
"""数据验证器"""

user_schema = {
"type": "object",
"required": ["user_id", "username", "email"],
"properties": {
"user_id": {"type": "integer", "minimum": 1},
"username": {"type": "string", "minLength": 3, "maxLength": 50},
"email": {"type": "string", "format": "email"},
"phone": {"type": "string", "pattern": "^1[3-9]\\\\d{9}$"},
"roles": {
"type": "array",
"items": {"type": "string", "enum": ["admin", "user", "moderator"]},
"default": ["user"]
}
}
}

order_schema = {
"type": "object",
"required": ["order_id", "user_id", "items", "total_amount"],
"properties": {
"order_id": {"type": "string"},
"user_id": {"type": "integer", "minimum": 1},
"items": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["product_id", "quantity", "price"],
"properties": {
"product_id": {"type": "integer"},
"quantity": {"type": "integer", "minimum": 1},
"price": {"type": "number", "minimum": 0}
}
}
},
"total_amount": {"type": "number", "minimum": 0}
}
}

@staticmethod
def validate_user(data):
try:
validate(instance=data, schema=DataValidator.user_schema)
return True, None
except ValidationError as e:
return False, e.message

@staticmethod
def validate_order(data):
try:
validate(instance=data, schema=DataValidator.order_schema)
return True, None
except ValidationError as e:
return False, e.message

# 使用示例
# 有效用户数据
valid_user = {
"user_id": 1001,
"username": "zhangsan",
"email": "zhangsan@example.com"
}
valid, error = DataValidator.validate_user(valid_user)
print(f"用户验证: {'通过' if valid else f'失败: {error}'}")

# 有效订单数据
valid_order = {
"order_id": "ORD001",
"user_id": 1001,
"items": [
{"product_id": 1, "quantity": 2, "price": 99.99}
],
"total_amount": 199.98
}
valid, error = DataValidator.validate_order(valid_order)
print(f"订单验证: {'通过' if valid else f'失败: {error}'}")

第8章：错误处理和最佳实践

8.1 常见JSON错误类型

import json

# 1. JSONDecodeError – JSON格式错误
try:
json.loads('{"name": "张三"}') # 有效
json.loads('{"name": "张三",}') # 无效：尾部逗号
except json.JSONDecodeError as e:
print(f"JSONDecodeError: {e.msg}")
print(f" 行: {e.lineno}, 列: {e.colno}")

# 2. TypeError – 类型错误
try:
json.loads(123) # 需要字符串
except TypeError as e:
print(f"TypeError: {e}")

# 3. UnicodeDecodeError – 编码错误
try:
with open("invalid.json", 'rb') as f:
json.load(f)
except UnicodeDecodeError as e:
print(f"UnicodeDecodeError: {e}")

8.2 错误处理策略

import json

class JSONErrorHandler:
"""JSON错误处理器"""

def __init__(self):
self.error_count = 0
self.errors = []

def safe_loads(self, json_str):
"""安全解析JSON"""
try:
return json.loads(json_str), None
except json.JSONDecodeError as e:
self.error_count += 1
error_info = {
'type': 'JSONDecodeError',
'message': e.msg,
'line': e.lineno,
'column': e.colno
}
self.errors.append(error_info)
return None, error_info
except Exception as e:
self.error_count += 1
error_info = {'type': type(e).__name__, 'message': str(e)}
self.errors.append(error_info)
return None, error_info

# 使用示例
handler = JSONErrorHandler()

# 有效JSON
data, error = handler.safe_loads('{"name": "张三"}')
if data:
print(f"✓ 解析成功: {data}")

# 无效JSON
data, error = handler.safe_loads('{"name": "张三",}')
if error:
print(f"✗ 解析失败: {error}")

print(f"总错误数: {handler.error_count}")

8.3 性能优化最佳实践

import json
import time

# 1. 使用紧凑格式
data = {"users": [{"id": i, "name": f"user{i}"} for i in range(1000)]}

# 紧凑格式（更快，更小）
start = time.time()
compact = json.dumps(data, separators=(',', ':'))
compact_time = time.time() – start
compact_size = len(compact)

# 格式化（更慢，更大）
start = time.time()
formatted = json.dumps(data, indent=2)
formatted_time = time.time() – start
formatted_size = len(formatted)

print(f"紧凑格式: {compact_time:.4f}s, {compact_size}字节")
print(f"格式化: {formatted_time:.4f}s, {formatted_size}字节")
print(f"大小差异: {(1 – compact_size/formatted_size)*100:.1f}%")

8.4 安全性最佳实践

import json

# 1. 验证输入
def safe_json_loads(json_str):
"""安全的JSON解析"""
if not isinstance(json_str, str):
raise TypeError("输入必须是字符串")

if len(json_str) > 1024 * 1024: # 限制1MB
raise ValueError("JSON数据过大")

# 检查深度
depth = json_str.count('{') + json_str.count('[')
if depth > 100:
raise ValueError("JSON深度过大")

return json.loads(json_str)

# 2. 防止注入
def sanitize_json(json_str):
"""清理JSON字符串"""
# 移除危险字符
dangerous_chars = ['\\x00', '\\x01', '\\x02']
for char in dangerous_chars:
json_str = json_str.replace(char, '')
return json_str

8.5 可维护性最佳实践

import json
from typing import Any, Dict, Optional

def load_json_file(filepath: str, encoding: str = 'utf-8') –> Dict[str, Any]:
"""加载JSON文件"""
try:
with open(filepath, 'r', encoding=encoding) as f:
return json.load(f)
except FileNotFoundError:
raise FileNotFoundError(f"文件不存在: {filepath}")
except json.JSONDecodeError as e:
raise ValueError(f"JSON格式错误: {e.msg}")

def save_json_file(filepath: str, data: Dict[str, Any],
encoding: str = 'utf-8', indent: int = 2) –> None:
"""保存JSON文件"""
with open(filepath, 'w', encoding=encoding) as f:
json.dump(data, f, ensure_ascii=False, indent=indent)

# 使用示例
try:
config = load_json_file("config.json")
config["new_key"] = "new_value"
save_json_file("config.json", config)
except (FileNotFoundError, ValueError) as e:
print(f"错误: {e}")

8.6 实际生产场景应用

import json
import logging
from typing import Any, Dict, Optional
from datetime import datetime

class JSONService:
"""JSON服务"""

def __init__(self, logger: Optional[logging.Logger] = None):
self.logger = logger or logging.getLogger(__name__)
self.metrics = {
'total_requests': 0,
'successful_requests': 0,
'failed_requests': 0,
'total_bytes': 0
}

def parse_request(self, request_body: str) –> Optional[Dict[str, Any]]:
"""解析请求体"""
self.metrics['total_requests'] += 1

try:
data = json.loads(request_body)
self.metrics['successful_requests'] += 1
self.metrics['total_bytes'] += len(request_body)
return data
except json.JSONDecodeError as e:
self.metrics['failed_requests'] += 1
self.logger.error(f"JSON解析失败: {e.msg}")
return None
except Exception as e:
self.metrics['failed_requests'] += 1
self.logger.error(f"解析异常: {str(e)}")
return None

def create_response(self, data: Dict[str, Any],
status: str = "success") –> str:
"""创建响应"""
response = {
"status": status,
"data": data,
"timestamp": datetime.now().isoformat()
}
return json.dumps(response, ensure_ascii=False)

# 使用示例
service = JSONService()

# 有效请求
request1 = '{"user_id": 1001, "action": "login"}'
data = service.parse_request(request1)
if data:
response = service.create_response({"user_id": 1001})
print(f"响应: {response}")

# 无效请求
request2 = '{"user_id": 1001,}' # 无效JSON
data = service.parse_request(request2)
if not data:
print("请求解析失败")

print(f"指标: {service.metrics}")

第9章：综合实战案例

9.1 项目需求分析

需求：

9.2 数据模型设计

import json
from datetime import datetime
from decimal import Decimal
import uuid

class User:
"""用户模型"""

def __init__(self, user_id, username, email, phone=None, roles=None):
self.user_id = user_id
self.username = username
self.email = email
self.phone = phone
self.roles = roles or ["user"]
self.created_at = datetime.now()
self.is_active = True

def to_dict(self):
return {
"user_id": self.user_id,
"username": self.username,
"email": self.email,
"phone": self.phone,
"roles": sorted(list(self.roles)),
"created_at": self.created_at.isoformat(),
"is_active": self.is_active
}

@classmethod
def from_dict(cls, data):
"""从字典创建用户对象"""
user = cls(
user_id=data["user_id"],
username=data["username"],
email=data["email"],
phone=data.get("phone"),
roles=set(data.get("roles", ["user"]))
)
user.is_active = data.get("is_active", True)
if "created_at" in data:
user.created_at = datetime.fromisoformat(data["created_at"])
return user

class Product:
"""商品模型"""

def __init__(self, product_id, name, price, stock=0, categories=None, description=""):
self.product_id = product_id
self.name = name
self.price = price
self.stock = stock
self.categories = categories or set()
self.description = description
self.created_at = datetime.now()

def to_dict(self):
return {
"product_id": self.product_id,
"name": self.name,
"price": float(self.price),
"stock": self.stock,
"categories": sorted(list(self.categories)),
"description": self.description,
"created_at": self.created_at.isoformat()
}

@classmethod
def from_dict(cls, data):
"""从字典创建商品对象"""
product = cls(
product_id=data["product_id"],
name=data["name"],
price=Decimal(str(data["price"])),
stock=data.get("stock", 0),
categories=set(data.get("categories", [])),
description=data.get("description", "")
)
if "created_at" in data:
product.created_at = datetime.fromisoformat(data["created_at"])
return product

class OrderItem:
"""订单项"""

def __init__(self, product_id, product_name, quantity, price):
self.product_id = product_id
self.product_name = product_name
self.quantity = quantity
self.price = price

def to_dict(self):
return {
"product_id": self.product_id,
"product_name": self.product_name,
"quantity": self.quantity,
"price": float(self.price)
}

class Order:
"""订单模型"""

VALID_STATUSES = ["pending", "paid", "shipped", "delivered", "cancelled"]

def __init__(self, order_id, user_id, items, shipping_address):
self.order_id = order_id
self.user_id = user_id
self.items = items
self.shipping_address = shipping_address
self.status = "pending"
self.created_at = datetime.now()
self.updated_at = None
self.total_amount = self._calculate_total()

def _calculate_total(self):
"""计算订单总额"""
total = sum(item.price * item.quantity for item in self.items)
return total

def to_dict(self):
return {
"order_id": str(self.order_id),
"user_id": self.user_id,
"items": [item.to_dict() for item in self.items],
"shipping_address": self.shipping_address,
"status": self.status,
"total_amount": float(self.total_amount),
"created_at": self.created_at.isoformat(),
"updated_at": self.updated_at.isoformat() if self.updated_at else None
}

@classmethod
def from_dict(cls, data):
"""从字典创建订单对象"""
items = [
OrderItem(
product_id=item["product_id"],
product_name=item["product_name"],
quantity=item["quantity"],
price=Decimal(str(item["price"]))
)
for item in data["items"]
]

order = cls(
order_id=uuid.UUID(data["order_id"]),
user_id=data["user_id"],
items=items,
shipping_address=data["shipping_address"]
)
order.status = data.get("status", "pending")
order.total_amount = Decimal(str(data.get("total_amount", 0)))

if "created_at" in data:
order.created_at = datetime.fromisoformat(data["created_at"])
if "updated_at" in data and data["updated_at"]:
order.updated_at = datetime.fromisoformat(data["updated_at"])

return order

def update_status(self, new_status):
"""更新订单状态"""
if new_status not in self.VALID_STATUSES:
raise ValueError(f"无效的状态: {new_status}")
if new_status == self.status:
return False

self.status = new_status
self.updated_at = datetime.now()
return True

9.3 JSON Schema定义

# 用户Schema
user_schema = {
"type": "object",
"required": ["user_id", "username", "email"],
"properties": {
"user_id": {"type": "integer", "minimum": 1},
"username": {"type": "string", "minLength": 3, "maxLength": 50},
"email": {"type": "string", "format": "email"},
"phone": {"type": "string", "pattern": "^1[3-9]\\\\d{9}$"},
"roles": {
"type": "array",
"items": {"type": "string", "enum": ["admin", "user", "moderator"]},
"default": ["user"]
},
"is_active": {"type": "boolean", "default": True},
"created_at": {"type": "string", "format": "date-time"}
}
}

# 商品Schema
product_schema = {
"type": "object",
"required": ["product_id", "name", "price", "stock"],
"properties": {
"product_id": {"type": "integer", "minimum": 1},
"name": {"type": "string", "minLength": 1, "maxLength": 100},
"price": {"type": "number", "minimum": 0, "maximum": 1000000},
"stock": {"type": "integer", "minimum": 0},
"categories": {
"type": "array",
"items": {"type": "string"},
"default": []
},
"description": {"type": "string", "maxLength": 1000},
"created_at": {"type": "string", "format": "date-time"}
}
}

# 订单Schema
order_schema = {
"type": "object",
"required": ["order_id", "user_id", "items", "shipping_address", "status"],
"properties": {
"order_id": {"type": "string", "pattern": "^[0-9a-f-]{36}$"},
"user_id": {"type": "integer", "minimum": 1},
"items": {
"type": "array",
"minItems": 1,
"items": {
"type": "object",
"required": ["product_id", "product_name", "quantity", "price"],
"properties": {
"product_id": {"type": "integer"},
"product_name": {"type": "string", "minLength": 1},
"quantity": {"type": "integer", "minimum": 1},
"price": {"type": "number", "minimum": 0}
}
}
},
"shipping_address": {
"type": "object",
"required": ["name", "phone", "address", "city", "zipcode"],
"properties": {
"name": {"type": "string", "minLength": 1, "maxLength": 50},
"phone": {"type": "string", "pattern": "^1[3-9]\\\\d{9}$"},
"address": {"type": "string", "minLength": 1, "maxLength": 200},
"city": {"type": "string", "minLength": 1, "maxLength": 50},
"zipcode": {"type": "string", "pattern": "^\\\\d{6}$"}
}
},
"status": {"type": "string", "enum": ["pending", "paid", "shipped", "delivered", "cancelled"]},
"total_amount": {"type": "number", "minimum": 0},
"created_at": {"type": "string", "format": "date-time"},
"updated_at": {"type": "string", "format": "date-time"}
}
}