【AWS】Lambda Powertoolsを布教したい！

こんにちは。メディア研究開発センター（M研）の嘉田です。
最近は専らAWS（とダイエット）のことばかり考えています。

早速ですが、みなさんはAWS Lambdaを使っていますか？
私が開発に携わっている社内向け文字起こしサービス「YOLO」では、EC2中心のレガシーなアーキテクチャーから、LambdaやFargateを活用したサーバレスアーキテクチャへと徐々に移行しています。
YOLOについては下記の記事をご覧いただければと思います。

今回はLambdaの開発時におすすめのライブラリ、AWS Lambda Powertoolsについて語りたいと思います。この記事を読んだ人がみんな使いたくなりますように…。

Lambda Powertoolsとは？

Lambda Powertoolsは、Lambda開発のためのユーティリティライブラリで、構造化ログ出力、X-Rayによるトレース、CloudWatchのカスタムメトリクスの作成などを簡単に実装できる優れものです。
Powertoolsの主要機能についてはAWSのbuilders.flashの連載で丁寧に紹介されているので、興味のある方はこちらを一読することをおすすめします。
※ Powertoolsのバージョンは古いのでご注意ください。

AWS Lambda Powertools Python 入門 第 1 回 - 変化を求めるデベロッパーを応援するウェブマガジン | AWS

現在対応している言語はPython・Java・TypeScript・.NETです。この記事のサンプルコードは全てPythonになります。

ちなみにYOLOでは、メイン機能である音声認識まわりをLambdaに移行するタイミングでPowertoolsを導入しました（2023年3月、下記記事もご覧ください）。

導入して感じたメリットはざっとこんな感じでしょうか。

• 構造化ログが簡単に生成される

• バリデーションを自前で実装しなくて良くなる

• API Gatewayとの統合がいろいろ楽になる

それぞれ後ほど紹介したいと思いますが、とにかくよしなにやってくれるのです。

インストール方法と使い方

導入はとても簡単で、Lambdaレイヤーかpipから使えます。

AWSコンソールでサクッと使う

「AWSレイヤー」にPowertoolsのレイヤーが用意されています。使いたいLambda関数でこのレイヤー追加するだけです！

ローカルでの開発時

LambdaのコードをローカルPCで開発・テストすることもありますよね。
ローカルPCではpipインストールすることでPowertoolsが使えます。

pip install aws-lambda-powertools

後述のValidationを使うときには下記になります。

pip install aws-lambda-powertools[validation]

デプロイ時

YOLOでは下記のように運用しており、ともにAWS CDKでデプロイしています。

コンテナイメージで作るLambda：requirements.txtに追加してDockerfileでpip install
通常のLambda：こちらを参考に提供されているレイヤーを追加
※ 最新のレイヤーはこちらから確認できます。

何ができる？

それでは、実際に使っている機能を紹介していきます。
今回紹介するのはPowertoolsの一部の機能に過ぎないので、興味のある方はぜひ公式ドキュメントなど見てみてください。

Logger

まずは構造化ログの出力を行うLoggerについてです。
一番簡単に取り入れやすいものかと思います。

⚙ 設定

まず、下記2つの環境変数を設定します。

LOG_LEVEL：ログレベル
POWERTOOLS_SERVICE_NAME：サービス名

サービス名はログ中に含まれるもので、定義しない場合はservice_undefinedとなります。

👨‍💻 基本的な使い方

サンプルコードは下記の通りです。
Pythonのlogging.Loggerクラスを継承しており、logger.info("hoge")のように慣れ親しんだ書き方で使えます。
inject_lambda_contextデコレータを使うことで、Lambda Context、コールドスタート情報を出力できます。余談ですがコールドスタート情報はLambda Insightsでも取得できて、コールドスタートにかかる時間や割合などが簡単に確認できて楽しいです。
log_eventパラメータをTrueにすると、入力イベント（{”key1”: “value1”, …}）が自動的に出力されます（機密情報の出力にはご注意を）。logger.info(event)などと書かなくてよくなりますね。

from aws_lambda_powertools import Logger

logger = Logger()

@logger.inject_lambda_context(log_event=True)
def lambda_handler(event, context):
    logger.info("hoge")
    logger.info({"key": "value"})
    return {"statusCode": 200}

実行すると、Cloudwatch Logsでは下記のようなログが確認できます。フォーマットなど指定せずともきれいにログ出力してくれるので最高です。

inject_lambda_contextデコレータをなくした場合、下記のようなログになります。

📁 別ファイルでのログ出力

lambda_handlerを定義したファイル以外でもログを出力したい場合、下記のようにchildパラメータを使うことで共通のLoggerをコード全体で利用できます。別ファイルにすることはよくあるので助かります。

# app.py
from aws_lambda_powertools import Logger
from child import child_test

logger = Logger()

def lambda_handler(event, context):
    logger.info("parent!")
    child_test()
    return {"statusCode": 200}

# child.py
from aws_lambda_powertools import Logger

logger = Logger(child=True)

def child_test():
    logger.info("child!")

出力結果は下記のようになります。

🔍 pytest使用時

Lambdaコードの単体テストにpytestを使っていますが、inject_lambda_contextデコレータを使用したコードをテストする場合、ダミーのLambda Contextを渡さないとうまくいきません。そこで、下記のように必要な最低限の情報を渡してやります。

from dataclasses import dataclass

import pytest
from app import lambda_handler

@pytest.fixture
def lambda_context():
    @dataclass
    class LambdaContext:
        function_name: str = "test"
        memory_limit_in_mb: int = 128
        invoked_function_arn: str = (
            "arn:aws:lambda:ap-northeast-1:000000000:function:test"
        )
        aws_request_id: str = "52fdfc07-2182-154f-163f-5f0f9a621d72"

    return LambdaContext()

def test_lambda_handler(lambda_context):
    test_event = {"test": "event"}
    lambda_handler(test_event, lambda_context)

また、pytest実行時にPowertoolsのログをきれいに表示するには、下記のように実行します。

POWERTOOLS_LOG_DEDUPLICATION_DISABLED="1" pytest -o log_cli=1

Loggerはこれ以外にも、構造化ログにキーを追加したり、ログサンプリングにより一定割合でDEBUGログレベルを有効にするなどもできるようです。非常に多機能ですね。

---

Validation

次に入力イベントとレスポンスのValidationです。脱自作しましょう！

👨‍💻 基本的な使い方

デコレータを使用する方法と関数を使用する方法があるので、それぞれ紹介します。

• validatorデコレータ

サンプルコードは下記の通りです。
schemas.pyに入力イベント・レスポンスの定義を記述しています（後述）。
ここでは入力イベント、レスポンスともにバリデーション対象としていますが、どちらかだけでも問題ありません。
無効な値が入ってきた場合には、SchemaValidationErrorが発生します。
次に紹介するvalidate関数を使う方法ではレスポンスのバリデーションはできないので、レスポンスのバリデーションがしたい場合はこちらを使うと良いかと思います。

import schemas
from aws_lambda_powertools.utilities.validation import validator

@validator(inbound_schema=schemas.INPUT, outbound_schema=schemas.OUTPUT)
def lambda_handler(event, context):
    return {"statusCode": 200}

• validate関数

サンプルコードは下記の通りです。
validate関数はlambda_handler内に限らず、好きなところで記述できます。
YOLOではデコレータではなくこちらで実装しています。lambda_handler関数の中でSchemaValidationErrorをキャッチして、処理を分岐させたかったからです（レスポンスのバリデーションは不要だと判断しました）。

import schemas
from aws_lambda_powertools.utilities.validation import SchemaValidationError, validate

def lambda_handler(event, context):
    try:
        validate(event=event, schema=schemas.INPUT)
        return {"statusCode": 200}
    except SchemaValidationError:
        do_something_with()
        raise

📝 入力イベントの定義

入力イベント・レスポンスの定義は下記のようにJSON Schemaで記述します。慣れないうちは面倒かもしれませんが、バリデーションのためのコードを長々書くよりは良いかと思います。

# schemas.py
INPUT = {
    "$schema": "http://json-schema.org/draft-07/schema",
    "type": "object",
    "examples": [{"path": "test.wav", "num": 1}],
    "required": ["path"],
    "properties": {
        "path": {"type": "string", "pattern": "[a-z0-9-_/]+\.wav"},
        "num": {"type": "integer", "minimum": 2, "maximum": 6},
    },
}

この定義に対して、{"path": "test.csv", "num": 2} という入力をしたとき、下記のようにエラーが出力されます。pathのパターンが違うとしっかり教えてくれていますね。

[ERROR] SchemaValidationError: Failed schema validation. Error: data.path must match pattern [a-z0-9-_/]+\.wav, Path: ['data', 'path'], Data: test.csv

ちなみにJSON Schemaの書き方は下記などを参考にしています。

JSON Schemaのすゝめ - Qiita

JSON Schema

✉️ envelope

envelopeパラメータを使うと、入力イベントの一部だけをバリデーションすることもできます。
これはEventBridgeやSNSの入力イベントのように多数の項目が渡ってくる場合に使えるかと思います。もちろんvalidatorデコレータでも使えます。

import schemas
from aws_lambda_powertools.utilities.validation import SchemaValidationError, validate

def lambda_handler(event, context):
    try:
        validate(event=event, schema=schemas.INPUT, envelope="detail")
        return {"statusCode": 200}
    except SchemaValidationError:
        raise

# 入力イベント例
{
    "id": "cdc73f9d-aea9-11e3-9d5a-835b769c0d9c",
    "detail-type": "Scheduled Event",
    "source": "aws.events",
    "account": "123456789012",
    "time": "1970-01-01T00:00:00Z",
    "region": "us-east-1",
    "resources": ["arn:aws:events:us-east-1:123456789012:rule/ExampleRule"],
    "detail": {
		# INPUTではdetailの中の項目だけ定義するのでOK
        "instance_id": "i-042dd005362091826",
        "region": "us-east-2",
    },
}

---

Event Handler

次にEvent Handlerです。
API Gateway・ALB・Lambda Function URLsの利用時にはぜひご検討ください。

👨‍💻 基本的な使い方

サンプルコードはREST APIのものです（Lambdaプロキシ統合である必要があります）。
FlaskやFastAPIのようにデコレータでルーティングを記述します。1つのLambdaで複数のAPIを記述することも簡単にできます（重量級Lambdaを生み出すのは微妙かと思いますが…）。
入力イベントはapp.current_eventで受け取れて、JSONデシリアライズされた辞書を受け取ることも、JSON文字列をそのまま受け取ることもできます。実際に使ってはいませんが、クエリ文字列も受け取れるようです。
レスポンスも、辞書型のオブジェクトをリターンすると自動でJSONシリアライズされるので、json.dumps()などは不要です。全てよしなにやってくれます。

from aws_lambda_powertools import Logger
from aws_lambda_powertools.event_handler import APIGatewayRestResolver
from aws_lambda_powertools.logging import correlation_paths
from aws_lambda_powertools.utilities.typing import LambdaContext

logger = Logger()
app = APIGatewayRestResolver()

@app.get("/hoge")
def get_todos():
    # JSONデシリアライズ
    event_dict = app.current_event.json_body
    # JSON文字列のまま
    event_str = app.current_event.body
    # クエリ文字列取得
    hoge_id = app.current_event.get_query_string_value(name="id", default_value="")
    return {"message": "test"}


@logger.inject_lambda_context(
    log_event=True, correlation_id_path=correlation_paths.API_GATEWAY_REST
)
def lambda_handler(event: dict, context: LambdaContext) -> dict:
    return app.resolve(event, context)

また、@app.get("/hoge/<id>")のようにして、動的ルーティングもできるようです。

🌐 CORS

CORSConfigクラスとcorsパラメータを使って、CORSの設定も簡単にできます。
オリジンがallow_originで指定した値と一致するときには自動で必要なヘッダーがレスポンスに含まれるので、レスポンスヘッダーを定義する手間から開放されます。

cors_config = CORSConfig(allow_origin="https://example.com", max_age=300)
app = APIGatewayRestResolver(cors=cors_config)

ただ、複数オリジンには対応していません…（参考）。長くなりそうなのでここでは書きませんが、OPTIONSメソッドを使って試行錯誤しているところです。

⚠️ 例外処理まわり

not_foundデコレータを使って、ルーティングが一致しない場合のレスポンスをカスタマイズすることや、exception_handlerデコレータを使って特定の例外が発生したときのレスポンスをカスタマイズできます。YOLOでは上記で紹介したValidationで例外が発生した場合などに利用しています。

@app.not_found
def handle_not_found_errors(ex: NotFoundError) -> Response:
    logger.info(f"Not Found Route: {app.current_event.path}")
    return Response(
        status_code=404,  # ステータスコードも自由に設定できる
        content_type=content_types.TEXT_PLAIN,
        body="Not Found Error.",
    )

@app.exception_handler(SchemaValidationError)
def handle_invalid_input(ex: SchemaValidationError) -> Response:
    return Response(
        status_code=400,
        content_type=content_types.TEXT_PLAIN,
        body="Invalid Input.",
    )

カスタマイズだけでなく、一般的なHTTPエラー（400、401、404、500）を簡単に返すためのクラスも用意されています。至れり尽くせりです。

---

Event Source Data Classes

最後に、Event Source Data Classesです。
Lambdaは様々なAWSサービスと連携することができますが、連携するサービスによって様々なフォーマットの入力イベントが生成されます。YOLOでもSNS、EventBridge、S3トリガーなどのサービスからLambdaを呼び出しており、それぞれサンプルリクエストとにらめっこしながらコードを書いていました。
そんなときに役立つのがEvent Source Data Classesで、一般的なイベント型に対する型ヒントやコード補完などを提供してくれます。

👨‍💻 基本的な使い方

入力イベントを各データクラスのコンストラクタに渡す方法と、event_sourceデコレータを使用する方法があります。下記はSNSの例です。

• データクラス

from aws_lambda_powertools.utilities.data_classes import SNSEvent

def lambda_handler(event, context):
    event = SNSEvent(event)
    for record in event.records:
        message = record.sns.message
        subject = record.sns.subject
        do_something_with(subject, message)

• event_sourceデコレータ

from aws_lambda_powertools.utilities.data_classes import SNSEvent, event_source

@event_source(data_class=SNSEvent)
def lambda_handler(event: SNSEvent, context):
    for record in event.records:
        message = record.sns.message
        subject = record.sns.subject
        do_something_with(subject, message)

VS Codeではこのように候補が出ます。地味に便利でした。

おわりに

ということで、Lambda Powertoolsの紹介でした。
自分の備忘録も兼ねて書き始めた記事でしたが、どこかの誰かに布教できていると万々歳です。
今後TracerやMetricsといった機能も使ってみたいと思います。

（メディア研究開発センター・嘉田紗世）