You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Pydantic 2.0:如何隐藏AWS SQS模型字段并用于计算字段?

问题描述

使用以下版本:

pydantic = "^2.0.2"
pydantic-settings = "^2.0.1"

需要为AWS SQS消息创建Pydantic BaseModel,要求调用print(model.model_dump_json())导出JSON时,输入字段message被隐藏。

现有未隐藏字段的实现代码功能正常,但导出JSON时message变量仍会显示:

from pydantic import BaseModel, computed_field
import json
from datetime import datetime

class SQSMessage(BaseModel):

    message: dict

    @property
    def sub_msg(self) -> dict:
        return json.loads(self.message["Message"])

    @computed_field
    @property
    def message_id(self) -> str:
        return self.message["MessageId"]

    @computed_field
    @property
    def bucket(self) -> str:
        return self.sub_msg["Records"][0]["s3"]["bucket"]["name"]

    @computed_field
    @property
    def key(self) -> str:
        return self.sub_msg["Records"][0]["s3"]["object"]["key"]

    @computed_field
    @property
    def event_datetime(self) -> datetime:
        _dt = self.sub_msg["Records"][0]["eventTime"]
        if "Z" in _dt:
            return datetime.fromisoformat(_dt.replace("Z", ""))
        else:
            return datetime.fromisoformat(_dt)

示例运行结果中message字段会被包含在输出的JSON里:

msg = {
        "Type": "Notification",
        "MessageId": "0d69f2aa-3384-5435-b75a-a9102074b9a3",
        "TopicArn": "arn:aws:sns:us-east-1:12345678910:example-sns-topic",
        "Subject": "Amazon S3 Notification",
        "Message": '{"Records":[{"eventVersion":"2.1","eventSource":"aws:s3","awsRegion":"us-east-1",'
                   '"eventTime":"2022-10-07T11:46:55.304Z","eventName":"ObjectCreated:Put",'
                   '"userIdentity":{"principalId":"AWS:AROARSMOKBLH3KNNAYOF2:ABCD12345@SOMEDOMAIN.com"},'
                   '"requestParameters":{"sourceIPAddress":"159.53.46.223"},'
                   '"s3":{"s3SchemaVersion":"1.0","configurationId":"tf-s3-topic-20220908215027661200000001",'
                   '"bucket":{"name":"app-bucket_name",'
                   '"ownerIdentity":{"principalId":"A1GXDFZQ55BKMX"},'
                   '"arn":"arn:aws:s3:::app-bucket_name"},'
                   '"object":{"key":"export/raw/v1/example.csv",'
                   '"size":0,'
                   '"eTag":"c258a1bafcdc3e550a13265867e8e5f4","sequencer":"00634011AF3640E697"}}}]}',
    }
model = SQSMessage(message=msg)
print(model.model_dump_json())
>>> {"message":{"Type":"Notification","MessageId":"0d69f2aa-3384-5435-b75a-a9102074b9a3","TopicArn":"arn:aws:sns:us-east-1:12345678910:example-sns-topic","Subject":"Amazon S3 Notification","Message":"{\"Records\":[{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventTime\":\"2022-10-07T11:46:55.304Z\",\"eventName\":\"ObjectCreated:Put\",\"userIdentity\":{\"principalId\":\"AWS:AROARSMOKBLH3KNNAYOF2:ABCD12345@SOMEDOMAIN.com\"},\"requestParameters\":{\"sourceIPAddress\":\"159.53.46.223\"},\"s3\":{\"s3SchemaVersion\":\"1.0\",\"configurationId\":\"tf-s3-topic-20220908215027661200000001\",\"bucket\":{\"name\":\"app-bucket_name\",\"ownerIdentity\":{\"principalId\":\"A1GXDFZQ55BKMX\"},\"arn\":\"arn:aws:s3:::app-bucket_name\"},\"object\":{\"key\":\"export/raw/v1/example.csv\",\"size\":0,\"eTag\":\"c258a1bafcdc3e550a13265867e8e5f4\",\"sequencer\":\"00634011AF3640E697\"}}}]}'"},"message_id":"0d69f2aa-3384-5435-b75a-a9102074b9a3","bucket":"app-bucket_name","key":"export/raw/v1/example.csv","event_datetime":"2022-10-07T11:46:55.304000"}

尝试使用PrivateAttr隐藏字段,但导致无法在computed_field中引用该字段,且导出JSON为空:

from pydantic import BaseModel, computed_field, PrivateAttr
import json
from datetime import datetime

class SQSMessage(BaseModel):
    _message: dict = PrivateAttr(default_factory=dict)

    @property
    def sub_msg(self) -> dict:
        return json.loads(self._message["Message"])

    @computed_field
    @property
    def message_id(self) -> str:
        return self._message["MessageId"]

    # 其余computed_field实现省略...

model = SQSMessage(_message=msg)
print(model.model_dump_json())
>>> {}

另外尝试通过Field(hidden=True)和json_schema_extra配置,也无法隐藏message字段:

from pydantic import BaseModel, computed_field, Field
import json
from datetime import datetime

class SQSMessage(BaseModel):
    class Config:
        @staticmethod
        def json_schema_extra(schema: dict, _):
            props = {}
            for k, v in schema.get('properties', {}).items():
                if not v.get("hidden", False):
                    props[k] = v
            schema["properties"] = props

    message: dict = Field(hidden=True)

    # 其余property和computed_field实现省略...

请问在Pydantic 2中该如何实现既能在模型内部引用message字段,又能在导出JSON时隐藏它的需求?


解决方案

在Pydantic 2中,有两种简单有效的方法可以实现该需求:

方法一:使用Field(exclude=True)

直接给message字段添加exclude=True参数,该字段会被排除在model_dump()和model_dump_json()的输出之外,但模型内部依然可以正常访问该字段。

修改后的代码:

from pydantic import BaseModel, computed_field, Field
import json
from datetime import datetime

class SQSMessage(BaseModel):
    message: dict = Field(exclude=True)

    @property
    def sub_msg(self) -> dict:
        return json.loads(self.message["Message"])

    @computed_field
    @property
    def message_id(self) -> str:
        return self.message["MessageId"]

    @computed_field
    @property
    def bucket(self) -> str:
        return self.sub_msg["Records"][0]["s3"]["bucket"]["name"]

    @computed_field
    @property
    def key(self) -> str:
        return self.sub_msg["Records"][0]["s3"]["object"]["key"]

    @computed_field
    @property
    def event_datetime(self) -> datetime:
        _dt = self.sub_msg["Records"][0]["eventTime"]
        if "Z" in _dt:
            return datetime.fromisoformat(_dt.replace("Z", ""))
        else:
            return datetime.fromisoformat(_dt)

运行示例代码后,输出的JSON中将不再包含message字段,只保留计算出的字段:

{"message_id":"0d69f2aa-3384-5435-b75a-a9102074b9a3","bucket":"app-bucket_name","key":"export/raw/v1/example.csv","event_datetime":"2022-10-07T11:46:55.304000"}

方法二:在model_dump_json()时指定exclude参数

如果不想修改字段定义,可以在调用导出方法时手动指定要排除的字段:

print(model.model_dump_json(exclude={"message"}))

这种方式更灵活,适合需要根据不同场景决定是否排除字段的情况。


内容的提问来源于stack exchange,提问作者Jenobi

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.11 13:17:34