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
相关产品推荐
相关产品推荐

