官方 json 文档（白名单/未知字段/序列化）

重要前提：

Python 内置 json 没有原生参数实现「白名单过滤 / 禁止未知字段」。 遇到 JSON 里多余的 key，json.loads() 默认直接全部读进 dict，不警告、不报错。

需要用 object_hook / object_pairs_hook 手动实现白名单、未知字段校验。 内置 json 不能直接序列化 dataclass、Enum，需要钩子处理。

1.核心函数速览

函数	作用
json.loads(s)	JSON 字符串 → Python dict/list
json.load(fp)	文件对象 → Python dict/list
json.dumps(obj)	Python 对象 → JSON 字符串
json.dump(obj, fp)	Python 对象 → 写入文件

序列化dumps关键参数

json.dumps(
    obj,
    skipkeys=False,
    ensure_ascii=True,
    allow_nan=True,   # JSON标准不支持NaN/Inf；生产建议 allow_nan=False
    default=None,     # 【重点】无法序列化的对象调用此回调
    indent=None,
)

default：当 json 不知道怎么序列化对象（dataclass、Enum），进入这个函数，你返回 dict/list；无法处理就抛 TypeError。

反序列化loads关键参数

json.loads(
    s,
    object_hook=None,        # 解析完每一个json object(dict)调用，返回值替换dict
    object_pairs_hook=None,  # 优先级更高，接收 [(k,v),…] 键值对列表，可检测重复key
)

object_hook 递归执行，嵌套对象也会进入钩子。

反序列化：处理未知字段（两种模式）

模式1：严格模式，遇到未知字段直接抛异常（推荐领域模型）

只允许白名单 key，一旦出现不在白名单的字段，直接抛出异常，防止混入底层坐标、关节等非法字段。

import json

WHITELIST = {"object", "destination", "max_retries"}

def strict_hook(d: dict):
    unknown_keys = set(d.keys()) - WHITELIST
    if unknown_keys:
        raise ValueError(f"存在未允许的字段: {unknown_keys}")
    return d

json_str = '''
{
    "object": "red_block",
    "destination": "bin_a",
    "max_retries": 3,
    "x": 0.12,
    "y": 0.23
}
'''

# x、y属于未知字段，触发报错
try:
    data = json.loads(json_str, object_hook=strict_hook)
except ValueError as e:
    print(e)

输出结果：

模式2：过滤模式，自动丢弃未知字段

多余 key 直接扔掉，只保留白名单内 key，不抛异常。

import json

WHITELIST = {"object", "destination", "max_retries"}

def filter_hook(d: dict):
    return {k: v for k, v in d.items() if k in WHITELIST}

json_str = '''
{
    "object": "red_block",
    "destination": "bin_a",
    "max_retries": 3,
    "x": 0.12
}
'''

data = json.loads(json_str, object_hook=filter_hook)
print(data)  # {'object': 'red_block', 'destination': 'bin_a', 'max_retries': 3}

输出结果：

注意：object_hook 递归作用，如果是嵌套 JSON，内层对象也会走该钩子，内层也要配置白名单。

序列化：dataclass/Enum的处理

内置 json 不能直接序列化 dataclass、Enum，直接 dumps 会报TypeError。

方式A:使用default钩子

from dataclasses import dataclass
from enum import StrEnum
import json

class ObjKind(StrEnum):
    RED_BLOCK = "red_block"

@dataclass
class PickTask:
    object: ObjKind
    destination: str
    max_retries: int = 2

def serialize_default(o):
    if isinstance(o, PickTask):
        # 序列化白名单：只输出指定字段
        return {
            "object": o.object.value,
            "destination": o.destination,
            "max_retries": o.max_retries
        }
    if isinstance(o, StrEnum):
        return o.value
    raise TypeError(f"无法序列化类型 {type(o)}")

task = PickTask(object=ObjKind.RED_BLOCK, destination="bin_a")
out = json.dumps(task, default=serialize_default, indent=2)
print(out)

代码分析：

入口：json.dumps(task, default=serialize_default, indent=2)

对象：task = PickTask(object=ObjKind.RED_BLOCK, destination="bin_a")

task 内部：

• object = <ObjKind.RED_BLOCK 枚举实例>
• destination = "bin_a"
• max_retries = 2（取默认值）

整体执行链路：

• 调用 json.dumps(task, default=serialize_default, indent=2)
• json 库尝试直接序列化顶层对象 task（PickTask 实例）
• json 原生不认识 PickTask 类型 → 触发调用你传入的 default=serialize_default(o)，此时参数o = task
• 进入serialize_default(o)函数
  • 判断 isinstance(o, PickTask) → True
  • 手动构造字典：

    {
        "object": o.object.value,   # o.object是枚举，.value取出"red_block"字符串
        "destination": o.destination,
        "max_retries": o.max_retries
    }
    

  • 返回这个普通 Python 字典
• json.dumps 拿到返回的普通 dict，字典里面的值全是基础类型（字符串、int），json 库可以直接处理
• indent=2 做格式化换行，生成 json 字符串
• 赋值给 out 变量，print 打印输出

方式B：dataclass.asdict（）先转字典再序列化

from dataclasses import dataclass, asdict
import json

@dataclass
class PickTask:
    object: str
    destination: str
    max_retries: int = 2

t = PickTask(object="red_block", destination="bin_a")
d = asdict(t)
# 序列化时手动过滤白名单
allow = {"object", "destination", "max_retries"}
filtered = {k:d[k] for k in allow}
s = json.dumps(filtered)

输出结果：

asdict()会递归转换所有字段，如果 dataclass 混入底层坐标字段，会全部输出；需要手动白名单裁剪

dataclass完整反序列化（标准库原生，无第三方）

json.loads出来只是普通 dict，不会自动变回 dataclass 实例，需要写类方法做转换，同时做白名单校验。

from dataclasses import dataclass
from enum import StrEnum
import json

class ObjKind(StrEnum):
    RED_BLOCK = "red_block"

@dataclass
class PickTask:
    object: ObjKind
    destination: str
    max_retries: int = 2

    @classmethod
    def from_dict(cls, d: dict):
        whitelist = {"object", "destination", "max_retries"}
        bad = set(d.keys()) - whitelist
        if bad:
            raise ValueError(f"不允许字段：{bad}")
        return cls(
            object=ObjKind(d["object"]),
            destination=d["destination"],
            max_retries=int(d.get("max_retries", 2))
        )

json_text = '''{"object":"red_block","destination":"bin_a","max_retries":2}'''
raw = json.loads(json_text)
task = PickTask.from_dict(raw)
print(task)

输出结果：

代码链路分析：

1. json_text 是 JSON 字符串。
2. json.loads(json_text)：json 库把字符串解析成普通 Python 字典 raw

raw = {
    "object":"red_block",
    "destination":"bin_a",
    "max_retries":2
}

1. 调用类方法 PickTask.from_dict(raw)，d 就是上面这个字典。

进入 from_dict(cls, d: dict)：

whitelist = {"object", "destination", "max_retries"}
bad = set(d.keys()) - whitelist

• whitelist：允许存在的 key 集合。
• set(d.keys()) 拿到输入字典所有 key；做集合减法，找出不在白名单里的字段放到bad。

如果 JSON 多传了比如 "foo":"123"，bad = {"foo"}，直接抛 ValueError: 不允许字段：{'foo'}，拒绝非法多余字段，防止脏数据灌入。

如果没有非法字段，往下执行：

return cls(
    object=ObjKind(d["object"]),
    destination=d["destination"],
    max_retries=int(d.get("max_retries", 2))
)

• ObjKind(d["object"])：字符串 "red_block" → 转回枚举实例 ObjKind.RED_BLOCK

如果 json 里 object 传个乱七八糟字符串，这里直接抛 ValueError，枚举构造器校验。

• destination=d["destination"]：直接拿字符串。
• d.get("max_retries",2)：字典没有 max_retries 时，给默认值 2；再包一层int()强制转成整数，防止 json 传字符串数字 "max_retries":"2"。

cls(...) 等价于 PickTask(...)，生成 PickTask 实例返回给task变量。

print(task) 输出 dataclass 打印结果：

官方行为的坑点（必看）

1. 多余字段静默接收：json.loads遇到不在预期的 key 不会报错，必须自己用object_hook/ 业务 from_dict 校验。
2. dataclass、Enum 不能直接 dumps，必须default钩子或者先转 dict。
3. JSON 重复 key：{"a":1,"a":2}，python json 默认取最后一个值，不抛异常；要用object_pairs_hook检测重复 key。
4. bool是int子类，json 解析不会做类型校验；max_retries:true会读到True(1)，需要业务层校验类型。
5. allow_nan=True默认允许输出NaN/Inf，这不符合 JSON 标准，对外接口建议设置allow_nan=False。

---

对比第三方库（了解）

• pydantic v2：extra="forbid" 原生禁止未知字段，自动序列化 dataclass/model，不用手写钩子。
• dataclasses‑json：专门为 dataclass 做 json 序列化。

标准库 json 没有以上能力，全部需要手动实现。

精读分析task_json.py

代码：

"""JSON 边界：合法 JSON -> PickAndPlaceTask；字段白名单 + 拒绝低层字段（第 1 周 D3）。"""

from __future__ import annotations

import json
from typing import Any

from .errors import DomainError, ErrorCode
from .task import FORBIDDEN_LOW_LEVEL_FIELDS, DestinationType, ObjectType, PickAndPlaceTask

_ALLOWED_FIELDS: frozenset[str] = frozenset({"object", "destination", "max_retries"})


def parse_task(raw: str) -> PickAndPlaceTask:
    """把外部 JSON 解析为任务对象；任何非法输入都抛 DomainError。"""
    try:
        data = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise DomainError(
            ErrorCode.INVALID_JSON, details={"line": exc.lineno, "col": exc.colno}
        ) from exc

    if not isinstance(data, dict):
        raise DomainError(ErrorCode.INVALID_JSON, details={"reason": "顶层必须是 JSON 对象"})

    _reject_low_level(data)
    _reject_unknown(data)
    _reject_missing(data)
    return _build(data)


def _reject_low_level(data: dict[str, Any]) -> None:
    hit = sorted(FORBIDDEN_LOW_LEVEL_FIELDS & set(data))
    if hit:
        raise DomainError(ErrorCode.FORBIDDEN_LOW_LEVEL_FIELD, details={"forbidden_fields": hit})


def _reject_unknown(data: dict[str, Any]) -> None:
    unknown = sorted(set(data) - _ALLOWED_FIELDS)
    if unknown:
        raise DomainError(ErrorCode.INVALID_TASK, details={"unknown_fields": unknown})


def _reject_missing(data: dict[str, Any]) -> None:
    missing = sorted(_ALLOWED_FIELDS - set(data))
    if missing:
        raise DomainError(ErrorCode.MISSING_FIELD, details={"missing_fields": missing})


def _build(data: dict[str, Any]) -> PickAndPlaceTask:
    try:
        obj = ObjectType(data["object"])
        destination = DestinationType(data["destination"])
        retries_value = data["max_retries"]
        if isinstance(retries_value, bool) or not isinstance(retries_value, int):
            raise ValueError("max_retries 必须是整数")
        max_retries = retries_value
    except (KeyError, TypeError, ValueError) as exc:
        raise DomainError(ErrorCode.INVALID_TASK, details={"reason": "字段取值非法"}) from exc
    try:
        return PickAndPlaceTask(object=obj, destination=destination, max_retries=max_retries)
    except ValueError as exc:
        raise DomainError(ErrorCode.INVALID_TASK, details={"reason": str(exc)}) from exc

核心目标：

外部传入的原始 JSON 字符串，层层校验，只有全部规则通过，才构造PickAndPlaceTask；只要任意一项不满足，抛出业务域异常 DomainError，携带错误码 + 详情，方便上层日志、返回错误给调用方。

模块依赖：

• DomainError, ErrorCode：自定义业务异常，每个错误对应枚举错误码
• FORBIDDEN_LOW_LEVEL_FIELDS：禁止传入的底层字段集合（防止外部直接下发底层机器人控制字段）
• ObjectType / DestinationType：StrEnum 枚举，对应物体、目标位置
• PickAndPlaceTask：dataclass 任务实体

_ALLOWED_FIELDS = 允许的字段：object / destination / max_retries

完整调用链路 parse_task (raw)

入口函数 parse_task(raw: str) -> PickAndPlaceTask

raw JSON字符串
    ↓
1. json.loads(raw)
    └─JSON语法错 → 抛DomainError(INVALID_JSON)，附带行号、列号
    ↓拿到data(Python对象)
2. 判断是否为dict
    └─顶层是数组/数字/字符串 → DomainError(INVALID_JSON):顶层必须是对象
    ↓
3. _reject_low_level(data) 拦截底层禁止字段
    └─如果data里出现FORBIDDEN_LOW_LEVEL_FIELDS中的字段
       → DomainError(FORBIDDEN_LOW_LEVEL_FIELD) 返回被命中的字段列表
    ↓
4. _reject_unknown(data) 白名单校验，拒绝未知字段
    └─存在不在_ALLOWED_FIELDS的key
       → DomainError(INVALID_TASK) 返回unknown_fields列表
    ↓
5. _reject_missing(data) 校验必填字段全部存在
    └─允许的字段里，有哪些没传
       → DomainError(MISSING_FIELD) 返回missing_fields列表
    ↓全部校验通过
6. _build(data) 做类型转换、业务校验，构造任务实例
    ↓
    6‑1 ObjectType(data["object"]) 字符串转枚举；非法枚举值抛异常
    6‑2 DestinationType(data["destination"])
    6‑3 max_retries强校验：不能是bool，必须是int
        >⚠️注意：Python里bool是int子类，`isinstance(True, int) → True`，所以必须单独把bool拦掉
    └─上面任意一步(KeyError/TypeError/ValueError) → 包装为DomainError(INVALID_TASK)
    ↓
    6‑4 实例化PickAndPlaceTask，构造失败同样包装业务异常
    ↓返回合法PickAndPlaceTask实例

每个子函数职责

1._reject_low_level(data)

hit = sorted(FORBIDDEN_LOW_LEVEL_FIELDS & set(data))

集合交集：看输入字典是否包含底层高危字段。

业务含义：上层业务任务接口，不允许客户端直接下发机器人底层控制字段，防止越权操控底层。 只要命中任意一个禁止字段，直接报错。

2._reject_unknown(data)

unknown = sorted(set(data) - _ALLOWED_FIELDS)

输入 key 集合 减去 白名单集合，剩下就是未知多余字段，直接拒绝。 等价之前 from_dict 里面的白名单校验，抽成独立函数。

3._reject_missing(data)

missing = sorted(_ALLOWED_FIELDS - set(data))

白名单全部字段都要求必须提供，少一个直接报 MISSING_FIELD。

和上一版 from_dict 不一样：上一版max_retries允许缺失、给默认值；这里逻辑是：max_retries 是必填字段，JSON 必须显式传入，不能靠内部默认填充。

重点差异： 旧 from_dict：d.get("max_retries",2) → 缺了就填默认 当前代码：_reject_missing，max_retries属于_ALLOWED_FIELDS，缺失直接报错，JSON 必须带上。

4._build (data)

负责把干净、经过字段校验的 dict，转为强类型对象。

• ObjectType(...) / DestinationType(...)：字符串还原枚举，非法值抛 ValueError
• max_retries特殊防护：

if isinstance(retries_value, bool) or not isinstance(retries_value, int):
    raise ValueError("max_retries 必须是整数")

Python 坑：isinstance(True, int)返回 True，如果只写isinstance(x,int)，会把 True/False 误判成整数。所以必须先排除 bool。 JSON 布尔值true/false进来，直接拒绝，不允许当重试次数。

所有原生异常全部用raise DomainError(...) from exc链式包装，保留原始异常栈，上层捕获统一处理业务错误。

整体架构思想

先校验格式与字段，再做类型转换与对象构造

• 第一步：JSON 语法解析
• 第二步：结构校验（是不是对象）
• 第三步：安全拦截（禁止底层字段）
• 第四步：字段白名单（不能多）
• 第五步：字段完整性（不能少）
• 最后一步：类型转换 + 构造实体对象

原则：外部不可信输入，全部校验完成之后，才进入业务对象构造。不在构造过程中做校验。

测试：跑 test_task_json.py 四类输入：合法 / 空值（空串、null）/ 非法动作（未知枚举）/ 越权坐标（x/y/z、joint_angles）→ 确认 FORBIDDEN_LOW_LEVEL_FIELD ；

test_task_json.py初始源码：

"""task_json 解析测试（D3/D5）：合法 / 空值 / 非法动作 / 越权坐标。"""

from __future__ import annotations

import json

import pytest

from arm_agent.domain.errors import DomainError, ErrorCode
from arm_agent.domain.task import DestinationType, ObjectType, PickAndPlaceTask
from arm_agent.domain.task_json import parse_task


def test_valid_json_to_task() -> None:
    task = parse_task('{"object":"red_block","destination":"left_bin","max_retries":2}')
    assert task == PickAndPlaceTask(ObjectType.RED_BLOCK, DestinationType.LEFT_BIN, max_retries=2)


@pytest.mark.parametrize(
    "payload",
    [
        '{"object":"red_block","destination":"left_bin","max_retries":2,"joint_angles":[0,0,0,0,0,0]}',
        '{"object":"red_block","destination":"left_bin","max_retries":2,"x":0.1,"y":0.2,"z":0.3}',
        '{"object":"red_block","destination":"left_bin","max_retries":2,"velocity":1.0}',
    ],
)
def test_low_level_fields_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.FORBIDDEN_LOW_LEVEL_FIELD


def test_unknown_field_rejected() -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task('{"object":"red_block","destination":"left_bin","max_retries":2,"color":"red"}')
    assert exc_info.value.code is ErrorCode.INVALID_TASK
    assert "color" in exc_info.value.details["unknown_fields"]


@pytest.mark.parametrize("payload", ['{"object":"red_block"}', '{"destination":"left_bin"}'])
def test_missing_field_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.MISSING_FIELD


@pytest.mark.parametrize("payload", ["", "not json", "[]", "3"])
def test_invalid_json_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.INVALID_JSON


@pytest.mark.parametrize(
    "payload",
    [
        '{"object":"gold_block","destination":"left_bin","max_retries":2}',  # 未知目标
        '{"object":"red_block","destination":"middle_shelf","max_retries":2}',  # 未知容器
        '{"object":"red_block","destination":"left_bin","max_retries":true}',  # 布尔冒充整数
        '{"object":"red_block","destination":"left_bin","max_retries":-1}',  # 负数
    ],
)
def test_invalid_values_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.INVALID_TASK


def test_roundtrip() -> None:
    raw = '{"object":"red_block","destination":"left_bin","max_retries":2}'
    task = parse_task(raw)
    assert json.loads(raw) == task.to_dict()

整体测试清单：

• test_valid_json_to_task：正常合法 JSON，解析出来对象完全匹配预期
• test_low_level_fields_rejected：越权底层字段拦截测试，传入关节角、坐标、速度这类底层控制字段，必须报 FORBIDDEN_LOW_LEVEL_FIELD
• test_unknown_field_rejected：出现白名单以外的普通多余字段（color），报 INVALID_TASK，details 要带上非法字段名
• test_missing_field_rejected：必填字段缺一个，触发 MISSING_FIELD
• test_invalid_json_rejected：空字符串、乱文本、JSON 数组、数字，全部判定非法 JSON
• test_invalid_values_rejected：枚举不存在、布尔冒充 int、负数重试次数，全部拒绝
• test_roundtrip：闭环测试：JSON 字符串 → 解析成 task 对象 → to_dict()转字典，和原始输入字典相等，验证序列化 / 反序列化双向一致性

逐个函数拆解

1.test_valid_json_to_task

def test_valid_json_to_task() -> None:
    task = parse_task('{"object":"red_block","destination":"left_bin","max_retries":2}')
    assert task == PickAndPlaceTask(ObjectType.RED_BLOCK, DestinationType.LEFT_BIN, max_retries=2)

正向冒烟测试：合法输入，解析出来的 dataclass 实例，和直接构造的对象相等。

2. test_low_level_fields_rejected（参数化）

@pytest.mark.parametrize(
    "payload",
    [
        '{"object":"red_block","destination":"left_bin","max_retries":2,"joint_angles":[0,0,0,0,0,0]}',
        '{"object":"red_block","destination":"left_bin","max_retries":2,"x":0.1,"y":0.2,"z":0.3}',
        '{"object":"red_block","destination":"left_bin","max_retries":2,"velocity":1.0}',
    ],
)
def test_low_level_fields_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.FORBIDDEN_LOW_LEVEL_FIELD

3. test_unknown_field_rejected

def test_unknown_field_rejected() -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task('{"object":"red_block","destination":"left_bin","max_retries":2,"color":"red"}')
    assert exc_info.value.code is ErrorCode.INVALID_TASK
    assert "color" in exc_info.value.details["unknown_fields"]

多传普通无关字段color，不是底层高危字段，但不在允许字段白名单。

• 错误码 INVALID_TASK
• 额外断言：检查 details 里面确实把非法字段color带出来，方便日志排查。

对应源码：_reject_unknown()。

4.test_missing_field_rejected（参数化）

@pytest.mark.parametrize("payload", ['{"object":"red_block"}', '{"destination":"left_bin"}'])
def test_missing_field_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.MISSING_FIELD

只传部分必填字段，缺少其他必填 key。 _ALLOWED_FIELDS里三个字段全部强制必填，少任意一个触发MISSING_FIELD。

注意：这里max_retries不支持缺省默认，缺失直接报错，不能靠内部填充默认值。 对应源码：_reject_missing()。

5.test_invalid_json_rejected（参数化）

@pytest.mark.parametrize("payload", ["", "not json", "[]", "3"])
def test_invalid_json_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.INVALID_JSON

覆盖 4 种畸形输入：

1. 空字符串
2. 完全不是 JSON 文本
3. JSON 顶层是数组[]（要求顶层必须是对象 {}）
4. JSON 顶层是数字3

全部抛出INVALID_JSON。

对应源码：json.loads 捕获异常 + 判断isinstance(data, dict)。

6.test_invalid_values_rejected（参数化）

@pytest.mark.parametrize(
    "payload",
    [
        '{"object":"gold_block","destination":"left_bin","max_retries":2}',  # 未知目标
        '{"object":"red_block","destination":"middle_shelf","max_retries":2}',  # 未知容器
        '{"object":"red_block","destination":"left_bin","max_retries":true}',  # 布尔冒充整数
        '{"object":"red_block","destination":"left_bin","max_retries":-1}',  # 负数
    ],
)
def test_invalid_values_rejected(payload: str) -> None:
    with pytest.raises(DomainError) as exc_info:
        parse_task(payload)
    assert exc_info.value.code is ErrorCode.INVALID_TASK

全部是：字段 key 齐全，但字段内容非法：

1. gold_block：物体枚举不存在
2. middle_shelf：目标位置枚举不存在
3. max_retries:true：JSON 布尔，源码专门拦截 bool
4. max_retries:-1：负数重试次数

7. test_roundtrip 往返闭环测试

def test_roundtrip() -> None:
    raw = '{"object":"red_block","destination":"left_bin","max_retries":2}'
    task = parse_task(raw)
    assert json.loads(raw) == task.to_dict()

往返（round‑trip）： 原始 JSON 字符串 → parse_task 解析成对象 → 调用task.to_dict()序列化输出字典 拿原始 json.loads 的字典和 to_dict () 输出对比，必须相等。 作用：保证序列化、反序列化双向不会丢字段、不会变形。

隐含前提：PickAndPlaceTask 实现了.to_dict()方法，对应前面的序列化逻辑。