この記事はデジタルキューブグループエンジニアチームアドベントカレンダー2024 12月5日の記事です。

目的

apigateway + lambdaでバックエンドの開発をする際に、API リクエストのバリデーションにpowertools for aws lambdaを使っている

この構成でバリデーションが正しく動いているか確認する必要が出てきたため、本記事のような検証を行った

環境

python(3.11.x)
pydantic(v2.x)
- powertools for aws lambdaの一部機能に pydantic が使われているが、pydanticが部分的にしか使えないので大元のパッケージも利用する
powertools for aws lambda(v3.x)

バリデーションを行う際のコード

lambda本体側は以下の感じ

from aws_lambda_powertools.utilities.parser import parse, ValidationError
from models import Hoge # 各自pydanticなクラスを定義したものをimportする

def handler(event, context):
    try:
        request_parameter = parse(event["body"], Hoge)
    except ValidationError as e:
        # エラー時にlambdaが返すレスポンスやエラーログ出力などの処理

    # なにか後続の処理

任意のキーと必須のキー混合

from typing import Any
from pydantic import RootModel, model_validator

class Hoge(RootModel[dict[str, Any]]):
   # 基本的には全てのキー/バリューを許可、一部必須フィールドをチェックするためのvalidatorを定義
   @model_validator(mode="before")
   def required_field_validation(self: dict[str, Any]):
       keys = self.keys()
       
       # 必須キーのキー名と型を定義
       required = {
           "key1": [float],
           "key2": [str, int],
       }
       
       error_messages = []
       
       for key_name, types in required.items():
           if key_name not in keys:
               error_messages.append(f"{key_name} is required")
           elif type(self[key_name]) not in types:
               error_messages.append(f"{key_name} must be {' or '.join([str(t.__name__) for t in types])}")

       if error_messages:
           raise ValueError("\n".join(error_messages))

       return self

任意のキーすべてを受け入れる

apiのpostで以下のようなbodyを受け入れる場合

{
    "hoge": "hoge",
    "fuga": 1
}

以下のように定義

from typing import Any
from pydantic import RootModel

class Hoge(RootModel[dict[str, Any]]):
    pass

配列を受け入れる

apiのpostで以下のようなbodyを受け入れる場合

[
    {
        "hoge": "hoge",
        "fuga": 1
    },
    {
        "hoge": "hoge",
        "fuga": 2
    },
    {
        "hoge": "hoge",
        "fuga": 3
    },
]

以下のように定義

from typing import Any
from pydantic import RootModel

class Hoge(RootModel[list[dict[str, Any]]]):
    pass

任意のdictでないならたとえば以下でいいはず

from typing import Any
from pydantic import RootModel, BaseModel

class HogeDict(BaseModel):
    hoge: str
    fuga: int

class Hoge(RootModel[list[HogeDict]]):
    pass

その他

基本的にはよくこういう感じで使うのでメモ的に残しておく

from pydantic import BaseModel

class HogeProps(BaseModel):
    hoge: str
    fuga: int

class Hoge(BaseModel):
    props: HogeProps
    name: str
    age: int

想定されるリクエストボディ

{
    "name": "name",
    "age": 111,
    "props": {
        "hoge": "hoge",
        "fuga": 1
    }
}