Powertools for AWS Lambda (Python)を使うときに気をつけたいDynamoDB JSONのデシリアライズ
2024.08.02AWSのサーバーレス開発をするときに何かと便利なPowertools for AWS Lambda(以下、Powertools)ですが、DynamoDBストリームのイベントを扱うときに少しつまづいたので記録として残します
バージョン
aws-lambda-powertools : "2.36.0"
※ バージョンによって動作が異なる可能性があります
参考：Powertoolsについて
つまづいたことポイント
PowertoolsのDynamoDBStreamEventデータクラスを使用して、DynamoDBストリームをトリガーとするLambdaを作成中、eventに渡されるDynamoDB JSON形式を想定して辞書にアクセスしたらエラーになる
↑だとよくわからないと思うので実際のコードで見ると、
DynamoDBStreamEventの構造(例)
"event": {
"type": "<class 'aws_lambda_powertools.utilities.data_classes.dynamo_db_stream_event.DynamoDBStreamEvent'>",
"value": {
"Records": [
{
"eventID": "",
"eventName": "INSERT",
"eventVersion": "1.1",
"eventSource": "aws:dynamodb",
"awsRegion": "ap-northeast-1",
"dynamodb": {
"ApproximateCreationDateTime": 1722483841,
"Keys": {
"pk": {
"S": "pk_value"
}
},
"SequenceNumber": "",
"SizeBytes": 30,
"StreamViewType": "KEYS_ONLY"
},
"eventSourceARN":
}
]
}
}
ここの"Keys"要素から"pk"の値を取得する際、以下のような2重の辞書を考慮して
「pk = keys["pk"].["S"]」でアクセスするとエラーになる？
{"pk": {"S": "pk_value"}}
@event_source(data_class=DynamoDBStreamEvent)
def handler(
event: DynamoDBStreamEvent,
context,
):
for record in event.records:
keys = record.dynamodb.keys
pk = keys["pk"].["S"] # これだとエラー
pk = keys["pk"] # なぜかこれで取得できる
解決
powertoolsのソースを確認したところ、自動で「DynamoDB type(DynamoDB JSON)」を「python type」にデシリアライズする処理があるようでした
以下の28行目〜60行目のコメントで示唆してあります
つまり上記のLambdaのコードで言うと、「 keys=record.dynamodb.keys 」の時に
aws_lambda_powertools/utilities/data_classes/dynamo_db_stream_event.pyの、
@property
def keys(self) -> Optional[Dict[str, Any]]: # type: ignore[override]
"""The primary key attribute(s) for the DynamoDB item that was modified."""
return self._deserialize_dynamodb_dict("Keys")
_deserialize_dynamodb_dict("Keys")
def _deserialize_dynamodb_dict(self, key: str) -> Optional[Dict[str, Any]]:
"""Deserialize DynamoDB records available in `Keys`, `NewImage`, and `OldImage`
Parameters
----------
key : str
DynamoDB key (e.g., Keys, NewImage, or OldImage)
Returns
-------
Optional[Dict[str, Any]]
Deserialized records in Python native types
"""
dynamodb_dict = self._data.get(key)
if dynamodb_dict is None:
return None
return {k: self._deserializer.deserialize(v) for k, v in dynamodb_dict.items()}
が呼び出され、DynamoDB JSON形式のデータを扱いやすく変換してくれていたということです
{"pk": {"S": "pk_value"}} 変換前
{"pk": "pk_value"} 変換後 属性を指定しなくて良くなった👍
ちなみにPowertoolsの公式ドキュメントを確認しましたが、現時点でこの処理に関する説明が無いようでした・・・
終わりに
本件、生成AIに聞いてもpowertoolsのソースまでは把握してなかったみたいで、直接確認して初めてデシリアライズの処理が行われていることに気づきました
やはり1次情報は重要ですね
