JSON
JSON 全称:
JavaScript Object Notation
JSON 是一种轻量级的数据交换格式,常用于:
- Web API 返回数据
- 配置文件
- 前后端数据交换
- 程序之间传递数据
- 保存结构化数据
- 日志和任务消息
- 股票、商品、用户等数据存储
一段 JSON 数据:
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
"is_active": true,
"industry": null
}
它和 Python 字典很相似,但 JSON 是一种文本格式,不是 Python 对象。
Python 标准库自带 json 模块,不需要安装。
import json
常用函数:
| 函数 | 作用 |
|---|---|
json.dumps() |
Python 对象转 JSON 字符串 |
json.loads() |
JSON 字符串转 Python 对象 |
json.dump() |
Python 对象写入 JSON 文件 |
json.load() |
从 JSON 文件读取 Python 对象 |
可以记成:
带 s:处理字符串
不带 s:处理文件
也就是:
dumps → 转字符串
loads → 读字符串
dump → 写文件
load → 读文件
JSON 支持以下基本类型:
对象
数组
字符串
数字
布尔值
null
示例:
{
"name": "张三",
"age": 25,
"height": 1.75,
"is_active": true,
"skills": ["Python", "SQL"],
"address": {
"city": "成都",
"district": "武侯区"
},
"remark": null
}
使用大括号:
{
"code": "600519",
"name": "贵州茅台"
}
对应 Python:
{
"code": "600519",
"name": "贵州茅台",
}
使用方括号:
[
"600519",
"000001",
"300750"
]
对应 Python 列表:
[
"600519",
"000001",
"300750",
]
正确:
{
"name": "贵州茅台"
}
错误:
{
'name': '贵州茅台'
}
JSON 不支持单引号字符串。
JSON 使用小写:
true
false
Python 使用首字母大写:
True
False
JSON:
null
Python:
None
下面不是合法 JSON:
{
// 股票代码
"code": "600519"
}
标准 JSON 不支持:
// 单行注释
/* 多行注释 */
如果配置文件需要注释,可以考虑 TOML 或 YAML。
| Python 类型 | JSON 类型 |
|---|---|
dict |
object |
list、tuple |
array |
str |
string |
int、float |
number |
True |
true |
False |
false |
None |
null |
示例:
data = {
"code": "600519",
"price": 1500.5,
"count": 100,
"is_active": True,
"tags": ["白酒", "消费"],
"remark": None,
}
转换后:
{
"code": "600519",
"price": 1500.5,
"count": 100,
"is_active": true,
"tags": [
"白酒",
"消费"
],
"remark": null
}
| JSON 类型 | Python 类型 |
|---|---|
| object | dict |
| array | list |
| string | str |
| 整数 | int |
| 小数 | float |
| true | True |
| false | False |
| null | None |
JSON 数组转换为 Python 后始终是列表,不会转换为元组。
基本用法:
import json
stock = {
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
}
json_text = json.dumps(stock)
print(json_text)
print(type(json_text))
输出类似:
{"code": "600519", "name": "\u8d35\u5dde\u8305\u53f0", "price": 1500.5}
<class 'str'>
dumps() 返回的是字符串:
str
默认情况下,中文可能转换成 Unicode 转义序列:
\u8d35\u5dde\u8305\u53f0
设置:
ensure_ascii=False
完整写法:
json_text = json.dumps(
stock,
ensure_ascii=False,
)
print(json_text)
结果:
{"code": "600519", "name": "贵州茅台", "price": 1500.5}
处理中文数据时,通常建议始终设置:
ensure_ascii=False
使用:
indent=2
json_text = json.dumps(
stock,
ensure_ascii=False,
indent=2,
)
print(json_text)
结果:
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5
}
也可以使用四个空格:
indent=4
配置文件和调试输出通常使用:
indent=2
data = {
"name": "贵州茅台",
"price": 1500.5,
"code": "600519",
}
使用:
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
sort_keys=True,
)
结果会按照键名排序:
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5
}
适合:
- 测试结果比较
- 版本控制
- 稳定输出
- 生成签名前的规范数据
普通输出中会包含空格:
json.dumps(data)
结果:
{"code": "600519", "name": "贵州茅台"}
可以通过 separators 删除多余空格:
json_text = json.dumps(
data,
ensure_ascii=False,
separators=(",", ":"),
)
结果:
{"code":"600519","name":"贵州茅台"}
适合:
- 网络传输
- 消息队列
- 减少文件体积
- API 请求体
可读性优先时使用 indent,体积优先时使用 separators。
import json
json_text = """
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
"is_active": true,
"remark": null
}
"""
stock = json.loads(json_text)
print(stock)
print(type(stock))
结果:
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
"is_active": True,
"remark": None,
}
类型:
<class 'dict'>
访问字段:
print(stock["code"])
print(stock["name"])
print(stock["price"])
JSON 字符串:
json_text = """
[
{
"code": "600519",
"name": "贵州茅台"
},
{
"code": "000001",
"name": "平安银行"
}
]
"""
解析:
stocks = json.loads(json_text)
print(type(stocks))
结果:
<class 'list'>
遍历:
for stock in stocks:
print(
stock["code"],
stock["name"],
)
import json
stocks = [
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5,
},
]
写入文件:
with open(
"stocks.json",
"w",
encoding="utf-8",
) as file:
json.dump(
stocks,
file,
ensure_ascii=False,
indent=2,
)
生成的 stocks.json:
[
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5
}
]
注意:
- 使用
w模式会覆盖原文件 - 文本文件应明确指定
encoding="utf-8" - 中文数据建议设置
ensure_ascii=False
import json
with open(
"stocks.json",
"r",
encoding="utf-8",
) as file:
stocks = json.load(file)
print(stocks)
遍历:
for stock in stocks:
print(
stock["code"],
stock["name"],
stock["price"],
)
现代 Python 项目推荐使用 Path。
from pathlib import Path
import json
读取:
file_path = Path("stocks.json")
text = file_path.read_text(
encoding="utf-8",
)
stocks = json.loads(text)
写入:
text = json.dumps(
stocks,
ensure_ascii=False,
indent=2,
)
file_path.write_text(
text,
encoding="utf-8",
)
from pathlib import Path
import json
def read_json(
file_path: Path,
) -> object:
text = file_path.read_text(
encoding="utf-8",
)
return json.loads(text)
def write_json(
file_path: Path,
data: object,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
file_path.write_text(
text,
encoding="utf-8",
)
调用:
write_json(
Path("data/stocks.json"),
stocks,
)
把 Python 对象转换为字符串:
json_text = json.dumps(data)
返回:
str
把 Python 对象直接写入文件对象:
json.dump(
data,
file,
)
解析 JSON 字符串:
data = json.loads(json_text)
从文件对象中解析 JSON:
data = json.load(file)
dumps / loads 中的 s
可以理解为 string
错误 JSON:
json_text = """
{
"code": "600519",
"name": "贵州茅台",
}
"""
最后一个字段后面多了逗号。
解析:
data = json.loads(json_text)
会抛出:
json.JSONDecodeError
处理:
import json
try:
data = json.loads(json_text)
except json.JSONDecodeError as error:
print(f"JSON 解析失败:{error}")
try:
data = json.loads(json_text)
except json.JSONDecodeError as error:
print(f"错误信息:{error.msg}")
print(f"错误行号:{error.lineno}")
print(f"错误列号:{error.colno}")
print(f"字符位置:{error.pos}")
更友好的提示:
def parse_json(
json_text: str,
) -> object:
try:
return json.loads(json_text)
except json.JSONDecodeError as error:
raise ValueError(
"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列,"
f"{error.msg}"
) from error
from pathlib import Path
import json
def read_json(
file_path: Path,
) -> object:
try:
text = file_path.read_text(
encoding="utf-8",
)
except FileNotFoundError as error:
raise FileNotFoundError(
f"JSON 文件不存在:{file_path}"
) from error
except IsADirectoryError as error:
raise ValueError(
f"路径是目录,不是文件:{file_path}"
) from error
except PermissionError as error:
raise PermissionError(
f"无权读取文件:{file_path}"
) from error
except UnicodeDecodeError as error:
raise ValueError(
f"文件不是 UTF-8 编码:{file_path}"
) from error
try:
return json.loads(text)
except json.JSONDecodeError as error:
raise ValueError(
"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列,"
f"{error.msg}"
) from error
JSON 语法正确,不代表数据结构符合业务要求。
例如:
{
"code": "600519",
"name": "贵州茅台"
}
语法正确,但程序可能要求根节点必须是列表。
data = read_json(
Path("stocks.json")
)
if not isinstance(data, list):
raise ValueError(
"JSON 根节点必须是数组"
)
def parse_stock(
data: object,
) -> dict[str, object]:
if not isinstance(data, dict):
raise TypeError(
"股票数据必须是对象"
)
required_fields = {
"code",
"name",
"price",
}
missing_fields = (
required_fields
- data.keys()
)
if missing_fields:
raise ValueError(
"缺少字段:"
+ ", ".join(
sorted(missing_fields)
)
)
code = str(data["code"]).strip()
name = str(data["name"]).strip()
try:
price = float(data["price"])
except (
TypeError,
ValueError,
) as error:
raise ValueError(
f"价格格式错误:"
f"{data['price']!r}"
) from error
if (
len(code) != 6
or not code.isdigit()
):
raise ValueError(
f"股票代码无效:{code!r}"
)
if price < 0:
raise ValueError(
"股票价格不能小于零"
)
return {
"code": code,
"name": name,
"price": price,
}
def parse_stocks(
data: object,
) -> tuple[
list[dict[str, object]],
list[tuple[int, str]],
]:
if not isinstance(data, list):
raise TypeError(
"JSON 根节点必须是数组"
)
stocks: list[
dict[str, object]
] = []
errors: list[
tuple[int, str]
] = []
for index, item in enumerate(
data,
start=1,
):
try:
stock = parse_stock(item)
except (
TypeError,
ValueError,
) as error:
errors.append(
(
index,
str(error),
)
)
else:
stocks.append(stock)
return stocks, errors
调用:
raw_data = read_json(
Path("stocks.json")
)
stocks, errors = parse_stocks(
raw_data
)
for stock in stocks:
print(stock)
for index, error in errors:
print(
f"第 {index} 条错误:{error}"
)
下面这些类型不能被 json.dumps() 直接序列化:
setbytesPathdatetimedateDecimal- 自定义类对象
- 数据类对象
例如:
data = {
"codes": {
"600519",
"000001",
},
}
执行:
json.dumps(data)
会抛出:
TypeError: Object of type set is not JSON serializable
可以给 json.dumps() 提供自定义转换函数。
from pathlib import Path
from datetime import date, datetime
from decimal import Decimal
import json
def json_default(
value: object,
) -> object:
if isinstance(value, set):
return sorted(value)
if isinstance(value, Path):
return str(value)
if isinstance(
value,
(date, datetime),
):
return value.isoformat()
if isinstance(value, Decimal):
return str(value)
raise TypeError(
"无法序列化对象:"
f"{type(value).__name__}"
)
使用:
data = {
"codes": {
"600519",
"000001",
},
"file": Path("stocks.json"),
"update_time": datetime.now(),
"price": Decimal("1500.50"),
}
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
default=json_default,
)
print(json_text)
from datetime import datetime
import json
data = {
"update_time": datetime.now(),
}
直接转换会报错:
json.dumps(data)
转换为 ISO 8601 字符串:
json_text = json.dumps(
data,
default=lambda value: (
value.isoformat()
if isinstance(value, datetime)
else TypeError()
),
)
更推荐使用命名函数:
def json_default(
value: object,
) -> object:
if isinstance(value, datetime):
return value.isoformat()
raise TypeError(
f"不支持类型:"
f"{type(value).__name__}"
)
结果类似:
{
"update_time": "2026-07-10T17:30:00.123456"
}
解析时不会自动恢复成 datetime,而是普通字符串。
需要手动转换:
update_time = datetime.fromisoformat(
data["update_time"]
)
财务数据常用:
from decimal import Decimal
price = Decimal("1500.50")
直接序列化会报错。
可以转换为字符串:
def json_default(
value: object,
) -> object:
if isinstance(value, Decimal):
return str(value)
raise TypeError
结果:
{
"price": "1500.50"
}
也可以转换为浮点数:
return float(value)
但浮点数可能存在精度损失。
财务数据通常更建议保存为字符串:
{
"price": "1500.50"
}
读取后:
price = Decimal(
data["price"]
)
from dataclasses import (
asdict,
dataclass,
)
import json
定义数据类:
@dataclass
class Stock:
code: str
name: str
price: float
创建对象:
stock = Stock(
code="600519",
name="贵州茅台",
price=1500.5,
)
数据类不能直接传给:
json.dumps(stock)
先转成字典:
data = asdict(stock)
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
from dataclasses import (
asdict,
is_dataclass,
)
def json_default(
value: object,
) -> object:
if is_dataclass(value):
return asdict(value)
if isinstance(value, Path):
return str(value)
if isinstance(
value,
(date, datetime),
):
return value.isoformat()
if isinstance(value, Decimal):
return str(value)
if isinstance(value, set):
return sorted(value)
raise TypeError(
f"无法序列化类型:"
f"{type(value).__name__}"
)
使用:
json_text = json.dumps(
stock,
ensure_ascii=False,
indent=2,
default=json_default,
)
class Stock:
def __init__(
self,
code: str,
name: str,
price: float,
) -> None:
self.code = code
self.name = name
self.price = price
def to_dict(
self,
) -> dict[str, object]:
return {
"code": self.code,
"name": self.name,
"price": self.price,
}
使用:
stock = Stock(
"600519",
"贵州茅台",
1500.5,
)
json_text = json.dumps(
stock.to_dict(),
ensure_ascii=False,
indent=2,
)
class Stock:
def __init__(
self,
code: str,
name: str,
price: float,
) -> None:
self.code = code
self.name = name
self.price = price
@classmethod
def from_dict(
cls,
data: dict[str, object],
) -> "Stock":
return cls(
code=str(data["code"]),
name=str(data["name"]),
price=float(data["price"]),
)
读取:
data = json.loads(json_text)
stock = Stock.from_dict(data)
也可以继承:
json.JSONEncoder
from dataclasses import (
asdict,
is_dataclass,
)
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
import json
class AppJSONEncoder(
json.JSONEncoder
):
def default(
self,
value: object,
) -> object:
if is_dataclass(value):
return asdict(value)
if isinstance(value, Path):
return str(value)
if isinstance(
value,
(date, datetime),
):
return value.isoformat()
if isinstance(value, Decimal):
return str(value)
if isinstance(value, set):
return sorted(value)
return super().default(value)
使用:
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
cls=AppJSONEncoder,
)
简单项目中使用 default 函数通常更直观。
json.loads() 可以通过 object_hook 自定义字典解析过程。
import json
from datetime import datetime
JSON:
json_text = """
{
"code": "600519",
"update_time": "2026-07-10T17:30:00"
}
"""
自定义转换:
def object_hook(
data: dict[str, object],
) -> dict[str, object]:
update_time = data.get(
"update_time"
)
if isinstance(
update_time,
str,
):
try:
data["update_time"] = (
datetime.fromisoformat(
update_time
)
)
except ValueError:
pass
return data
解析:
data = json.loads(
json_text,
object_hook=object_hook,
)
此时:
type(data["update_time"])
是:
datetime
不要把所有看起来像日期的字符串都自动转换,最好使用明确字段名或类型标记。
默认情况下,JSON 小数解析为:
float
例如:
data = json.loads(
'{"price": 0.1}'
)
print(type(data["price"]))
结果:
<class 'float'>
财务数据可以转换为 Decimal:
from decimal import Decimal
data = json.loads(
'{"price": 0.1}',
parse_float=Decimal,
)
现在:
print(type(data["price"]))
结果:
<class 'decimal.Decimal'>
这在金额计算中很有用。
JSON 整数默认转换为:
int
可以自定义:
data = json.loads(
'{"quantity": 100}',
parse_int=str,
)
此时:
data["quantity"]
是字符串:
"100"
这种用法较少,一般保持默认即可。
Python 默认可能允许:
data = {
"value1": float("nan"),
"value2": float("inf"),
"value3": float("-inf"),
}
json_text = json.dumps(data)
print(json_text)
结果可能是:
{
"value1": NaN,
"value2": Infinity,
"value3": -Infinity
}
但这些值不属于严格 JSON 标准,其他系统可能无法解析。
推荐:
json.dumps(
data,
allow_nan=False,
)
遇到这些值时会抛出:
ValueError
需要提前清理数据。
import math
def clean_number(
value: object,
) -> object:
if (
isinstance(value, float)
and not math.isfinite(value)
):
return None
return value
复杂嵌套数据需要递归清理。
import math
from pathlib import Path
from datetime import date, datetime
from decimal import Decimal
def make_json_safe(
value: object,
) -> object:
if value is None:
return None
if isinstance(
value,
(str, int, bool),
):
return value
if isinstance(value, float):
if math.isfinite(value):
return value
return None
if isinstance(value, Decimal):
return str(value)
if isinstance(value, Path):
return str(value)
if isinstance(
value,
(date, datetime),
):
return value.isoformat()
if isinstance(value, dict):
return {
str(key): make_json_safe(
item
)
for key, item in value.items()
}
if isinstance(
value,
(list, tuple, set),
):
return [
make_json_safe(item)
for item in value
]
raise TypeError(
f"无法转换为 JSON:"
f"{type(value).__name__}"
)
使用:
safe_data = make_json_safe(
data
)
json_text = json.dumps(
safe_data,
ensure_ascii=False,
indent=2,
allow_nan=False,
)
JSON 对象的键必须是字符串。
Python 字典可以使用整数键:
data = {
1: "第一项",
2: "第二项",
}
序列化后:
{
"1": "第一项",
"2": "第二项"
}
重新解析后,键会变成字符串:
parsed = json.loads(
json.dumps(data)
)
print(parsed)
结果:
{
"1": "第一项",
"2": "第二项",
}
因此:
parsed != data
如果数据需要完整往返,JSON 字典键最好始终使用字符串。
如果字典包含 JSON 不支持的键:
data = {
("600519", "贵州茅台"): 1500,
}
默认会抛出:
TypeError
可以设置:
skipkeys=True
json_text = json.dumps(
data,
skipkeys=True,
)
非法键会被忽略。
重要数据不推荐使用 skipkeys=True,因为会静默丢失内容。
普通 JSON 数组:
[
{"code": "600519", "name": "贵州茅台"},
{"code": "000001", "name": "平安银行"}
]
JSON Lines 是每行一个独立 JSON 对象:
{"code":"600519","name":"贵州茅台"}
{"code":"000001","name":"平安银行"}
{"code":"300750","name":"宁德时代"}
常见扩展名:
.jsonl
.ndjson
JSON Lines 适合:
- 大量记录
- 日志
- 流式处理
- 逐条追加
- 单行错误隔离
- 数据导入导出
from pathlib import Path
import json
def write_json_lines(
file_path: Path,
records: list[
dict[str, object]
],
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
with file_path.open(
"w",
encoding="utf-8",
) as file:
for record in records:
line = json.dumps(
record,
ensure_ascii=False,
separators=(",", ":"),
)
file.write(line + "\n")
def append_json_line(
file_path: Path,
record: dict[str, object],
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
line = json.dumps(
record,
ensure_ascii=False,
separators=(",", ":"),
)
with file_path.open(
"a",
encoding="utf-8",
) as file:
file.write(line + "\n")
from collections.abc import Iterator
def read_json_lines(
file_path: Path,
) -> Iterator[
dict[str, object]
]:
with file_path.open(
"r",
encoding="utf-8",
) as file:
for line_number, line in enumerate(
file,
start=1,
):
clean_line = line.strip()
if not clean_line:
continue
try:
data = json.loads(
clean_line
)
except json.JSONDecodeError as error:
raise ValueError(
f"第 {line_number} 行 "
f"JSON 格式错误:"
f"{error.msg}"
) from error
if not isinstance(data, dict):
raise TypeError(
f"第 {line_number} 行 "
"必须是 JSON 对象"
)
yield data
使用:
for stock in read_json_lines(
Path("stocks.jsonl")
):
print(stock)
普通:
json.load(file)
或:
json.loads(text)
通常会一次性把整个 JSON 加载到内存。
对于非常大的 JSON 数组,可能占用大量内存。
可选方案:
- 改用 JSON Lines
- 使用数据库
- 使用流式 JSON 解析第三方库
- 按业务拆分成多个文件
- 使用 CSV、Parquet 等格式
如果数据由自己控制,优先考虑 JSON Lines:
每行一个对象
它比超大 JSON 数组更容易流式处理。
直接写入正式文件时,如果程序中途崩溃,文件可能损坏。
更安全的方式:
先写临时文件
→ 写入完成
→ 替换正式文件
from pathlib import Path
import json
import os
import tempfile
def safe_write_json(
file_path: Path,
data: object,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
allow_nan=False,
)
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=file_path.parent,
delete=False,
) as temp_file:
temp_file.write(text)
temp_file.flush()
temp_path = Path(
temp_file.name
)
try:
os.replace(
temp_path,
file_path,
)
except Exception:
temp_path.unlink(
missing_ok=True,
)
raise
适合:
- 配置文件
- 交易记录
- 重要数据结果
- 程序状态文件
- 不允许半写入的文件
假设原文件:
{
"theme": "dark",
"page_size": 20
}
读取、修改、写回:
from pathlib import Path
config_path = Path(
"config.json"
)
config = read_json(
config_path
)
if not isinstance(config, dict):
raise TypeError(
"配置文件根节点必须是对象"
)
config["page_size"] = 50
config["language"] = "zh-CN"
safe_write_json(
config_path,
config,
)
默认配置:
default_config = {
"theme": "light",
"page_size": 20,
"timeout": 10,
}
用户配置:
user_config = {
"theme": "dark",
"timeout": 30,
}
合并:
config = (
default_config
| user_config
)
结果:
{
"theme": "dark",
"page_size": 20,
"timeout": 30,
}
用户配置覆盖默认配置。
这只适用于浅层字典。
def merge_dicts(
base: dict[str, object],
override: dict[str, object],
) -> dict[str, object]:
result = base.copy()
for key, value in override.items():
old_value = result.get(key)
if (
isinstance(old_value, dict)
and isinstance(value, dict)
):
result[key] = merge_dicts(
old_value,
value,
)
else:
result[key] = value
return result
假设 API 返回:
response_data = {
"status_code": 0,
"data": [
{
"code": "600519",
"name": "贵州茅台",
},
{
"code": "000001",
"name": "平安银行",
},
],
}
检查状态:
status_code = response_data.get(
"status_code"
)
if status_code != 0:
raise RuntimeError(
f"接口返回错误:"
f"{status_code}"
)
获取数据:
records = response_data.get(
"data",
[],
)
if not isinstance(records, list):
raise TypeError(
"data 字段必须是数组"
)
遍历:
for record in records:
print(
record.get("code"),
record.get("name"),
)
原始数据:
response_data = {
"status_code": 0,
"data": [
{
"title": "要点一:公司完成股权收购",
"content": "公司完成目标企业股权收购。",
"update_date": "2026-07-01",
},
{
"title": "要点二:主营自动化仪表",
"content": "主要从事工业自动控制系统业务。",
"update_date": "2026-06-20",
},
],
}
转换函数:
def points_to_text(
response_data: dict[
str,
object
],
) -> str:
raw_items = response_data.get(
"data",
[],
)
if not isinstance(
raw_items,
list,
):
raise TypeError(
"data 必须是列表"
)
sections: list[str] = []
for index, item in enumerate(
raw_items,
start=1,
):
if not isinstance(
item,
dict,
):
continue
title = str(
item.get(
"title",
f"要点 {index}",
)
).strip()
content = str(
item.get(
"content",
"",
)
).strip()
update_date = str(
item.get(
"update_date",
"",
)
).strip()
lines = [title]
if content:
lines.append(content)
if update_date:
lines.append(
f"更新时间:{update_date}"
)
sections.append(
"\n".join(lines)
)
return "\n\n".join(
sections
)
使用:
text = points_to_text(
response_data
)
print(text)
按照股票代码去重:
stocks = [
{
"code": "600519",
"name": "贵州茅台",
},
{
"code": "000001",
"name": "平安银行",
},
{
"code": "600519",
"name": "贵州茅台",
},
]
def deduplicate_stocks(
stocks: list[
dict[str, object]
],
) -> list[
dict[str, object]
]:
result: list[
dict[str, object]
] = []
seen_codes: set[str] = set()
for stock in stocks:
code = str(
stock.get(
"code",
"",
)
)
if code in seen_codes:
continue
seen_codes.add(code)
result.append(stock)
return result
按照价格降序:
sorted_stocks = sorted(
stocks,
key=lambda stock: float(
stock.get(
"price",
0,
)
),
reverse=True,
)
按照代码:
sorted_stocks = sorted(
stocks,
key=lambda stock: str(
stock.get(
"code",
"",
)
),
)
排序后保存:
safe_write_json(
Path("sorted_stocks.json"),
sorted_stocks,
)
筛选以 00 或 60 开头的股票:
main_board_stocks = [
stock
for stock in stocks
if str(
stock.get(
"code",
"",
)
).startswith(
("00", "60")
)
]
筛选价格大于 100:
result = [
stock
for stock in stocks
if float(
stock.get(
"price",
0,
)
) > 100
]
from pathlib import Path
import csv
def json_to_csv(
json_path: Path,
csv_path: Path,
) -> None:
data = read_json(json_path)
if not isinstance(data, list):
raise TypeError(
"JSON 根节点必须是数组"
)
if not data:
csv_path.write_text(
"",
encoding="utf-8",
)
return
rows = [
row
for row in data
if isinstance(row, dict)
]
if not rows:
raise ValueError(
"JSON 中没有有效对象"
)
field_names: list[str] = []
seen_fields: set[str] = set()
for row in rows:
for key in row:
key_text = str(key)
if key_text in seen_fields:
continue
seen_fields.add(key_text)
field_names.append(key_text)
csv_path.parent.mkdir(
parents=True,
exist_ok=True,
)
with csv_path.open(
"w",
encoding="utf-8-sig",
newline="",
) as file:
writer = csv.DictWriter(
file,
fieldnames=field_names,
extrasaction="ignore",
)
writer.writeheader()
writer.writerows(rows)
def csv_to_json(
csv_path: Path,
json_path: Path,
) -> None:
with csv_path.open(
"r",
encoding="utf-8-sig",
newline="",
) as file:
reader = csv.DictReader(file)
rows = [
dict(row)
for row in reader
]
safe_write_json(
json_path,
rows,
)
CSV 中的所有字段默认都是字符串,需要根据业务继续转换类型。
config.json:
{
"database": {
"host": "localhost",
"port": 5432,
"name": "stock"
},
"output": {
"directory": "output",
"indent": 2
},
"debug": false
}
定义配置读取函数:
from pathlib import Path
def load_config(
file_path: Path,
) -> dict[str, object]:
data = read_json(file_path)
if not isinstance(data, dict):
raise TypeError(
"配置文件根节点必须是对象"
)
database = data.get(
"database"
)
if not isinstance(
database,
dict,
):
raise ValueError(
"缺少 database 配置"
)
required_fields = {
"host",
"port",
"name",
}
missing_fields = (
required_fields
- database.keys()
)
if missing_fields:
raise ValueError(
"数据库配置缺少字段:"
+ ", ".join(
sorted(missing_fields)
)
)
return data
不推荐含义不明确:
{
"c": "600519",
"n": "贵州茅台",
"p": 1500.5
}
推荐字段清晰:
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5
}
不推荐:
[
{
"price": 1500.5
},
{
"price": "11.5"
},
{
"price": null
}
]
字段类型不稳定会增加处理难度。
更推荐:
[
{
"price": 1500.5
},
{
"price": 11.5
}
]
缺失值是否允许,应在业务模型中明确。
长期使用的数据格式可以加入版本:
{
"version": 1,
"update_time": "2026-07-10T17:30:00",
"data": []
}
以后结构变化时,可以根据版本兼容处理。
错误字符串:
text = (
"{'code': '600519'}"
)
执行:
json.loads(text)
会失败,因为 JSON 必须使用双引号。
正确:
text = (
'{"code": "600519"}'
)
错误:
{
"code": "600519",
}
正确:
{
"code": "600519"
}
错误 JSON:
{
"active": True,
"remark": None
}
正确 JSON:
{
"active": true,
"remark": null
}
json.dumps(data)
中文会变成 Unicode 转义。
推荐:
json.dumps(
data,
ensure_ascii=False,
)
不推荐:
open(
"data.json",
"r",
)
推荐:
open(
"data.json",
"r",
encoding="utf-8",
)
错误:
json.dumps(
data,
file,
)
dumps() 不接收文件对象,它返回字符串。
正确写文件:
json.dump(
data,
file,
)
或者:
text = json.dumps(data)
file.write(text)
data = json.loads(text)
不能直接假设:
data["code"]
因为根节点可能是:
- 列表
- 字符串
- 数字
None
应先判断:
if not isinstance(data, dict):
raise TypeError(
"根节点必须是对象"
)
path.write_text(
json_text,
encoding="utf-8",
)
程序中途退出可能造成文件损坏。
重要数据建议使用安全写入。
{
"amount": 0.1
}
解析后是 float,计算可能出现精度误差。
可以:
json.loads(
text,
parse_float=Decimal,
)
或将金额保存为字符串。
JSON 只能直接保存有限的数据类型。
自定义类、日期、集合等需要先转换。
from dataclasses import (
asdict,
is_dataclass,
)
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
import json
import math
import os
import tempfile
def json_default(
value: object,
) -> object:
if is_dataclass(value):
return asdict(value)
if isinstance(value, Path):
return str(value)
if isinstance(
value,
(date, datetime),
):
return value.isoformat()
if isinstance(value, Decimal):
return str(value)
if isinstance(value, set):
return sorted(value)
raise TypeError(
f"无法序列化类型:"
f"{type(value).__name__}"
)
def read_json(
file_path: Path,
*,
parse_decimal: bool = False,
) -> object:
text = file_path.read_text(
encoding="utf-8",
)
try:
if parse_decimal:
return json.loads(
text,
parse_float=Decimal,
)
return json.loads(text)
except json.JSONDecodeError as error:
raise ValueError(
f"JSON 格式错误:"
f"第 {error.lineno} 行,"
f"第 {error.colno} 列,"
f"{error.msg}"
) from error
def write_json(
file_path: Path,
data: object,
*,
indent: int | None = 2,
safe: bool = True,
) -> None:
file_path.parent.mkdir(
parents=True,
exist_ok=True,
)
text = json.dumps(
data,
ensure_ascii=False,
indent=indent,
allow_nan=False,
default=json_default,
)
if not safe:
file_path.write_text(
text,
encoding="utf-8",
)
return
with tempfile.NamedTemporaryFile(
mode="w",
encoding="utf-8",
dir=file_path.parent,
delete=False,
) as temp_file:
temp_file.write(text)
temp_path = Path(
temp_file.name
)
try:
os.replace(
temp_path,
file_path,
)
except Exception:
temp_path.unlink(
missing_ok=True,
)
raise
原始文件 stocks.json:
[
{
"code": " 600519 ",
"name": "贵州茅台",
"price": 1500.5,
"industry": "白酒"
},
{
"code": "000001",
"name": "平安银行",
"price": 11.5,
"industry": "银行"
},
{
"code": "600519",
"name": "贵州茅台",
"price": 1500.5,
"industry": "白酒"
},
{
"code": "ABC",
"name": "错误数据",
"price": -10,
"industry": "未知"
}
]
定义数据类:
from dataclasses import (
asdict,
dataclass,
)
@dataclass(
slots=True,
frozen=True,
)
class Stock:
code: str
name: str
price: float
industry: str
解析单条数据:
def parse_stock(
value: object,
) -> Stock:
if not isinstance(
value,
dict,
):
raise TypeError(
"股票记录必须是对象"
)
try:
code = str(
value["code"]
).strip()
name = str(
value["name"]
).strip()
price = float(
value["price"]
)
industry = str(
value.get(
"industry",
"未知",
)
).strip()
except KeyError as error:
raise ValueError(
f"缺少字段:"
f"{error.args[0]}"
) from error
except (
TypeError,
ValueError,
) as error:
raise ValueError(
"股票字段类型错误"
) from error
if (
len(code) != 6
or not code.isdigit()
):
raise ValueError(
f"无效股票代码:"
f"{code!r}"
)
if price < 0:
raise ValueError(
"价格不能小于零"
)
return Stock(
code=code,
name=name,
price=price,
industry=industry,
)
批量处理:
def process_stock_data(
raw_data: object,
) -> tuple[
list[Stock],
list[tuple[int, str]],
]:
if not isinstance(
raw_data,
list,
):
raise TypeError(
"JSON 根节点必须是数组"
)
stocks: list[Stock] = []
errors: list[
tuple[int, str]
] = []
seen_codes: set[str] = set()
for index, value in enumerate(
raw_data,
start=1,
):
try:
stock = parse_stock(
value
)
except (
TypeError,
ValueError,
) as error:
errors.append(
(
index,
str(error),
)
)
continue
if stock.code in seen_codes:
continue
seen_codes.add(stock.code)
stocks.append(stock)
stocks.sort(
key=lambda stock: (
stock.price
),
reverse=True,
)
return stocks, errors
主程序:
def main() -> None:
input_path = Path(
"stocks.json"
)
output_path = Path(
"output/"
"clean_stocks.json"
)
raw_data = read_json(
input_path
)
stocks, errors = (
process_stock_data(
raw_data
)
)
output_data = [
asdict(stock)
for stock in stocks
]
write_json(
output_path,
output_data,
)
print(
f"成功保存:"
f"{len(stocks)} 条"
)
for index, error in errors:
print(
f"第 {index} 条错误:"
f"{error}"
)
if __name__ == "__main__":
main()
处理流程:
读取 JSON
→ 校验根节点
→ 校验每条记录
→ 清理字段
→ 删除重复代码
→ 按价格排序
→ 安全写入新文件
→ 输出错误记录
将 Python 字典转换为格式化 JSON 字符串,并正常显示中文。
参考答案:
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
读取 JSON 文件,并判断根节点是否为列表。
data = read_json(
Path("data.json")
)
if not isinstance(data, list):
raise TypeError(
"根节点必须是数组"
)
将集合转换为 JSON。
data = {
"codes": {
"600519",
"000001",
},
}
参考答案:
json_text = json.dumps(
data,
ensure_ascii=False,
indent=2,
default=lambda value: (
sorted(value)
if isinstance(value, set)
else TypeError()
),
)
用 Decimal 解析 JSON 小数。
from decimal import Decimal
data = json.loads(
'{"price": 1500.50}',
parse_float=Decimal,
)
把 JSON 数组中的股票按价格降序排列。
stocks = sorted(
stocks,
key=lambda stock: float(
stock.get(
"price",
0,
)
),
reverse=True,
)
筛选以 00 或 60 开头的股票代码。
result = [
stock
for stock in stocks
if str(
stock.get(
"code",
"",
)
).startswith(
("00", "60")
)
]
将多条数据以 JSON Lines 形式追加到文件。
for record in records:
append_json_line(
Path("stocks.jsonl"),
record,
)
导入:
import json
Python 对象转字符串:
text = json.dumps(
data,
ensure_ascii=False,
indent=2,
)
JSON 字符串转对象:
data = json.loads(text)
写入文件:
with open(
"data.json",
"w",
encoding="utf-8",
) as file:
json.dump(
data,
file,
ensure_ascii=False,
indent=2,
)
读取文件:
with open(
"data.json",
"r",
encoding="utf-8",
) as file:
data = json.load(file)
使用 Path:
path.write_text(
json.dumps(
data,
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
data = json.loads(
path.read_text(
encoding="utf-8",
)
)
捕获格式错误:
except json.JSONDecodeError as error:
自定义序列化:
json.dumps(
data,
default=json_default,
)
精确解析小数:
json.loads(
text,
parse_float=Decimal,
)
禁止非标准数值:
json.dumps(
data,
allow_nan=False,
)
JSON 处理主要分为四步:
序列化
反序列化
结构校验
业务校验
序列化:
Python 对象
→ JSON 字符串或文件
反序列化:
JSON 字符串或文件
→ Python 对象
需要重点记住:
dumps → Python 转 JSON 字符串
loads → JSON 字符串转 Python
dump → Python 写入 JSON 文件
load → 从 JSON 文件读取 Python
实际项目中应注意:
中文使用 ensure_ascii=False
格式化使用 indent=2
文件编码使用 UTF-8
解析后继续校验数据结构
金额考虑 Decimal
日期和自定义对象需要转换
大数据优先考虑 JSON Lines
重要文件使用安全写入
严格 JSON 使用 allow_nan=False
JSON 的本质是:
把结构化数据转换成通用文本
方便程序保存、交换和传输
但 JSON 语法正确只代表“格式合法”,并不代表“业务数据正确”。成熟的程序必须同时进行:
JSON 语法校验
+
数据类型校验
+
必需字段校验
+
业务规则校验