FastApi下Pydantic嵌套结构Json类型字段校验报错如何解决

错误原因 #

Pydantic内置的Json类型的校验逻辑是要求入参为可被反序列化的JSON字符串、字节或字节数组，会自动完成json.loads()操作转为Python对象。你在传入参数前已经对整段Kafka消息执行了json.loads()，导致otherData已经被解析为Python字典对象，不符合Json类型的入参要求，因此抛出类型错误。

解决方案 #

方案1：修改字段类型适配已反序列化结构（最常用）

如果你不需要保留otherData的原始JSON字符串，直接使用已解析后的Python结构，把Json类型替换为对应类型即可：

• 确定是字典结构：改为Optional[dict]
• 允许任意JSON兼容类型（字典、列表、字符串、数字、布尔、None）：可使用Optional[Any]，Pydantic v2也可以用内置的Optional[JsonValue]
  修改后的Schema示例：

from typing import Optional, Any

class DocumentInfoSchema(BaseModel):
    title: str
    type: str
    referenceId: int
    batchNoList: Optional[List]
    otherData: Optional[dict] # 替换原Json类型

class DocumentArticleSchema(BaseModel):
    type: str
    value: int
    accountType: Optional[AccountTypeEnums]
    accountId: Optional[int]
    otherData: Optional[dict] # 替换原Json类型

方案2：调整数据传入逻辑，统一交由Pydantic解析

如果你确实需要保留Json类型定义，不用提前对整个Kafka消息做json.loads()，直接调用Pydantic的parse_raw方法解析原始消息字符串即可，Pydantic会统一处理所有JSON解析和校验逻辑：
修改后的处理代码：

def process(self) -> bool:
    try:
        # 直接解析原始消息字符串，无需提前执行json.loads
        DocumentSchema.parse_raw(self._message)
        return self._process()

    except ValidationError as e:
        raise UnprocessableEntityException(e, self._topic)
    except ValueError as e:
        raise UnprocessableEntityException(e, self._topic)
    except Exception as e:
        raise UnprocessableEntityException(e, self._topic) 

方案3：手动序列化otherData字段（仅特殊场景使用）

如果你既需要提前解析整段消息，又不想修改原有Schema定义，可以在传给Schema之前手动把otherData字段序列化为JSON字符串，该方案会额外产生序列化/反序列化性能开销，无特殊需求不推荐使用：

def process(self) -> bool:
    try:
        msg_dict = json.loads(self._message)
        # 遍历嵌套结构，将otherData转为JSON字符串
        for doc in msg_dict.get("documentList", []):
            doc_info = doc.get("documentInfo", {})
            if doc_info.get("otherData"):
                doc_info["otherData"] = json.dumps(doc_info["otherData"])
            for article in doc.get("articleList", []):
                if article.get("otherData"):
                    article["otherData"] = json.dumps(article["otherData"])
        DocumentSchema(**msg_dict)
        return self._process()
    # 后续异常处理逻辑保持不变

可选优化 #

你现有代码中的Optional[List]没有指定列表内元素类型，建议补全为具体类型（比如Optional[List[str]]），否则Pydantic的校验效果会打折扣。

内容的提问来源于stack exchange，提问作者M a m a D