こんにちは。hayata-yamamoto です。
今まで、いくつかの記事を通して Powertools for AWS (Python) を用いた実装例を紹介してきました。今回は、今までの記事を踏まえた総論になるような記事にできればと考えています。「なぜ、Powertools を使うようになったのか」から始まり、実際にどのような運用をしているのかなどを共有できれば嬉しいです。
なぜ使うようになったのか
メンバーからの提案により採用が決まりました。採用したのはおおよそ2年前ほどで、当時は今よりもずっと運用している Lambda 関数が多く、Lambda 関数のテストを書くたびに、event のデータ型を調べて、辞書型でうまく扱って、と "Lambdaのことをどれくらい知っているか"がコード上で問われるような状態になっていました。
チーム内での認識を整えるためにその時やっていたのは、コメントにイベントの例を記載しておく、というもので、確かにキャッチアップの速度は解消してくれたのですが、コードを書く上では補完が効くわけでもなく、必ずしも実装しやすいとは言えない状況でした。（Optional Chaining ができる言語で書いていたらもう少し楽だったかもしれませんが）
この問題やその他、Lambdaを使うことで必要となる追加の知識をキャッチアップしやすくするためにメンバーが提案してくれて採用したのがきっかけでした。Powertools を使うことで Lambda のイベントをデータクラスで扱えたり、自分たちでイベント情報をカスタマイズした時も pydantic と組み合わせて利用できたりとさまざまな恩恵を得られるようになりました。
実際にどのように使っているか
ここでは、非同期呼び出しされる Lambda 関数を提供する際に弊社で採用している実装について紹介いたします。なお、各リソースとの統合用の Lambda 関数の実装については別記事をご確認いただければ幸いです。
具体的には、以下のようなシーケンスを想定しています。なお、弊社はAPIも FastAPI で Python製で Lambda 上で動いています。
実装はこんな感じでやってます
from aws_lambda_powertools.utilities.parser import BaseModel, event_parser
function_name = "" # 関数名
class HandlerNameAsyncEvent(BaseModel):
something_id: str
# 当該Lambda関数に対する関心をこのファイル内に留めておくことと、
# Lambdaの呼び出し方法が変わっても利用側が意識せず同じように呼び出すだけで良い状態を保つためです
# 例えば、Lambda の非同期呼び出しを使っていたが、SQS 経由で呼び出すようにする、など
# リリース後にいくつか提供方法を調整する場合があったりします
def request_handler_name_async(something: SomethingModel) -> None:
# SomethingModel は ORM のモデルです
# この関数をAPIのエンドポイント定義の中でimport して使ってもらってます
# boto3 の invoke 関数のラッパー（自社製）
# https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/lambda/client/invoke.html
invoke_lambda_function(
function_name=function_name,
_async=True,
payload=HandlerNameAsyncEvent(something_id=something.id).dict(),
)
@event_parser(model=HandlerNameAsyncEvent)
def handler(event: HandlerNameAsyncEvent, _: dict) -> None:
# 処理を追加
...
おわりに
もしご興味お持ちいただけた方は、ぜひカジュアル面談にお越しくださいませ！
Discussion