如何通过Pydantic自动解码AWS Lambda中S3事件的URL编码Key？

在AWS Lambda函数中使用Pydantic验证并类型化传入的S3 Event，事件结构如下：

{
  "Records": [
    {
      "s3": {
        "bucket": {
          "name": "my-bucket"
        },
        "object": {
          "key": "MYKEY%28CSV%29/XXXX.CSV"
        }
      }
    }
  ]
}

对应的Pydantic模型定义如下：

from pydantic import BaseModel

class ObjectInfo(BaseModel):
    key: str

class BucketInfo(BaseModel):
    name: str

class S3Schema(BaseModel):
    bucket: BucketInfo
    object: ObjectInfo

class Record(BaseModel):
    s3: S3Schema

class DeletionEvent(BaseModel):
    Records: list[Record]

def handler(event: dict, _):
    eventTyped = DeletionEvent(**event)
    return True

需要将key字段的URL编码值（如MYKEY%28CSV%29/XXXX.CSV）自动解码为原始值MYKEY(CSV)/XXXX.CSV，希望通过Pydantic自动完成解码，而非手动使用urllib.parse.unquote_plus处理。

解决方案 #

方法一：字段验证器直接处理 #

在ObjectInfo模型的key字段上添加Pydantic字段验证器，自动完成URL解码：

from pydantic import BaseModel, field_validator
from urllib.parse import unquote_plus

class ObjectInfo(BaseModel):
    key: str

    @field_validator('key')
    def decode_s3_key(cls, value):
        return unquote_plus(value)

class BucketInfo(BaseModel):
    name: str

class S3Schema(BaseModel):
    bucket: BucketInfo
    object: ObjectInfo

class Record(BaseModel):
    s3: S3Schema

class DeletionEvent(BaseModel):
    Records: list[Record]

def handler(event: dict, _):
    event_typed = DeletionEvent(**event)
    # 此时event_typed.Records[0].s3.object.key已解码为MYKEY(CSV)/XXXX.CSV
    return True

这种方式针对性强，仅对目标字段做解码，逻辑直观，无需额外类型定义。

方法二：自定义可复用的URL解码类型 #

如果多个字段或模型需要URL解码逻辑，可以自定义一个字符串类型，后续直接复用：

from pydantic import BaseModel, GetCoreSchemaHandler
from pydantic_core import core_schema
from urllib.parse import unquote_plus
from typing import Any

class UrlDecodedStr(str):
    @classmethod
    def __get_pydantic_core_schema__(cls, source_type: Any, handler: GetCoreSchemaHandler) -> core_schema.CoreSchema:
        # 基于基础字符串类型扩展，添加解码验证逻辑
        return core_schema.no_info_wrap_validator_function(
            cls.decode_value,
            handler(str),
        )

    @classmethod
    def decode_value(cls, value: str) -> str:
        return unquote_plus(value)

class ObjectInfo(BaseModel):
    key: UrlDecodedStr  # 直接使用自定义类型

# 其余BucketInfo、S3Schema等模型保持原定义不变

后续任何需要URL解码的字符串字段，只需将类型指定为UrlDecodedStr即可自动完成解码，适合多场景复用。

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