用 Python 建立机械臂任务领域模型-领域模型（上）： JSON 边界精读 + 4 类输入测试确认2026.8.12

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

重要前提：

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

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

1.核心函数速览

序列化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)

输出结果：

代码链路分析：

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

进入 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 打印结果：

官方行为的坑点（必看）

对比第三方库（了解）

• 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 种畸形输入：

全部抛出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 齐全，但字段内容非法：

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()方法，对应前面的序列化逻辑。