如何基于boto3的get_object响应构建适配AWS S3的Pydantic模型

模型修正说明 #

你当前编写的模型存在两处需要调整的地方，其中Body字段定义错误，LastModified定义基本准确可优化：

1. Body字段修正

• 原定义为str不符合实际返回类型：boto3的get_object返回的Body是botocore.response.StreamingBody流式对象，用于读取S3文件的二进制内容，可通过调用.read()方法获取字节数据。
• 由于StreamingBody不属于Pydantic默认支持的可序列化类型，需要在模型配置中开启arbitrary_types_allowed = True，否则会触发类型验证报错。
• 修正前需要先导入对应类型：from botocore.response import StreamingBody

2. LastModified字段优化

• 你当前使用的datetime.datetime类型可以正常兼容S3返回的带UTC时区的时间对象，不会出现验证错误。
• 若使用Pydantic V2，可替换为AwareDatetime类型强制要求时间携带时区信息，更贴合S3的返回特性，严谨性更高。

完整修正后示例 #

Pydantic V1 版本

import datetime
from botocore.response import StreamingBody
from pydantic import BaseModel

class S3GetObjectResponse(BaseModel):
    AcceptRanges: str
    Body: StreamingBody
    ContentLength: int
    ContentType: str
    ETag: str
    LastModified: datetime.datetime
    Metadata: dict
    ResponseMetadata: dict

    class Config:
        arbitrary_types_allowed = True

Pydantic V2 版本（更严谨）

from datetime import datetime
from botocore.response import StreamingBody
from pydantic import BaseModel, AwareDatetime
from pydantic.config import ConfigDict

class S3GetObjectResponse(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    
    AcceptRanges: str
    Body: StreamingBody
    ContentLength: int
    ContentType: str
    ETag: str
    LastModified: AwareDatetime
    Metadata: dict
    ResponseMetadata: dict

额外注意 #

StreamingBody类型无法直接序列化转JSON，如果你需要序列化该模型，需要提前调用Body.read()将内容读取为字节/字符串存储，或者自定义字段序列化规则处理该字段。

内容的提问来源于stack exchange，提问作者edmamerto