ここから始める Powertools for AWS Lambda (Python) - JMESPath 編 -

はじめに

こんにちは。
Penetrator プロダクト開発部で主にインフラを担当している遠藤です。

みなさん元気に Lambda Life を送っていますか?
Lambda 関数で JSON データを扱う際、複雑にネストされた構造から特定のデータを抽出したり、配列をフィルタリングしたりする処理に苦労した経験のある方もいらっしゃると思います。

そんな課題を解決してくれるのが、Powertools for AWS Lambda の JMESPath Functions 機能です。
今回はこの機能について、実際の使用イメージと共に紹介します。

なお、今後も Powertools for AWS Lambda の各種機能について継続的にブログを書いていく予定なので、興味のある方は楽しみにしていただけると嬉しいです。
※ 前回の Typing 編は コチラ。

Powertools for AWS Lambda とは

Powertools for AWS Lambda は、サーバーレスベストプラクティスの実装と開発者の生産性向上を支援する開発者ツールキットです (公式サイト)。
段階的に導入できる柔軟な設計により、必要な機能から順次採用することができます。

Python、TypeScript、Java、.NET で提供されており、AWS Well-Architected Serverless Lens に基づく推奨事項を簡単に実装できます。

JMESPath Functions 機能

JMESPath Functions 機能は、AWS Lambda イベントソースで一般的に見られるデータ形式のデシリアライゼーションを処理する組み込み JMESPath 関数を提供するユーティリティです (公式ドキュメント)。

主な特徴は以下の通りです:

• AWS Lambda イベント特化: JSON 文字列, Base64, Gzip 圧縮データなど、Lambda イベントソースで一般的なデータ形式のデシリアライゼーションを処理
• Powertools 全体での活用: API Gateway, Kinesis, CloudWatch Logs 等からのイベントを簡単にデコード・展開するために Powertools 内部で広く使用
• エンベロープ式での利用: JMESPath 式が許可される場所であればどこでも利用可能
• カスタム関数サポート: 特殊なバイナリ形式に対応するためのカスタム JMESPath 関数を追加可能

2025年7月現在、以下の組み込み関数が提供されています:

• query(): 任意の JMESPath 式でデータを検索・抽出
• powertools_json(): JSON 文字列をオブジェクトにデシリアライズ
• powertools_base64(): Base64 データをデコード
• powertools_base64_gzip(): Base64 デコード + Gzip 圧縮データ解凍

JMESPath

JMESPath は、JSON データに対するクエリ言語です (公式ドキュメント)。
AWS CLI、AWS Python SDK、Powertools for AWS Lambda でも使用されており、以下のような特徴があります:

• 直感的な記法: JSON の構造に沿った自然な表現
• 強力なフィルタリング: 条件に基づくデータ抽出
• 配列処理: map, filter, reduce 的な操作をサポート
• パイプライン処理: 複数の操作を連鎖させて実行

基本的な使用例:

# オブジェクトのプロパティ取得
user.name

# 配列の要素取得
users[0].email

# フィルタリング
users[?age > `20`].name

# 射影（全要素に対する操作）
users[*].name

Built-in Envelopes

Powertools for AWS Lambda では、主要な AWS Lambda イベントソースから簡単にペイロードを抽出できるよう、Built-in envelopes を提供しています。
エンベロープとは、JSON オブジェクトからデータを抽出するための JMESPath 式のことです。
AWS の各サービスは異なる形式でイベントを送信するため、実際に処理したいデータを取り出すには複雑な JMESPath 式が必要になることがあります。
主な Built-in envelopes には以下があります:

• API_GATEWAY_HTTP: HTTP API の body からペイロードを抽出
• API_GATEWAY_REST: REST API の body からペイロードを抽出
• CLOUDWATCH_EVENTS_SCHEDULED: detail からペイロードを抽出
• CLOUDWATCH_LOGS: gzip 解凍・Base64 デコードして logEvents からペイロードを抽出
• EVENTBRIDGE: detail からペイロードを抽出
• KINESIS_DATA_STREAM: Records を Base64 デコードしてペイロードを抽出
• S3_EVENTBRIDGE_SQS: Records の detail からペイロードを抽出
• S3_KINESIS_FIREHOSE: Records を Base64 デコードしてペイロードを抽出
• S3_SNS_KINESIS_FIREHOSE: Records を Base64 デコードしてペイロードを抽出
• S3_SNS_SQS: Records からペイロードを抽出
• S3_SQS: Records からペイロードを抽出
• SNS: Records からペイロードを抽出
• SQS: Records の body からペイロードを抽出

これらの built-in envelopes を使用することで、各 AWS サービス特有のイベント構造を意識することなく、簡潔にデータを抽出できます。

実践: Lambda 関数の構築・実装

ここからは、実際に Lambda 関数を構築・実装するイメージを紹介します。

※ プロジェクトの完全なコードは コチラ。

使用ツール	バージョン
aws-cdk-cli	2.1021.0
aws-cdk-lib	2.199.0
aws-cli	2.13.32
node	22.13.1
python	3.11.8
uv	0.6.14
jq	1.7

0. プロジェクトの初期化

Terminal

# プロジェクトディレクトリの作成
mkdir jmespath && \
cd jmespath

Terminal

# AWS CDK プロジェクトの初期化
cdk init app --language python

Terminal

# プロジェクトのセットアップ
source .venv/bin/activate && \
uv init && \
uv add --group infra -r requirements.txt && \
uv add --dev -r requirements-dev.txt && \
mkdir lambda && \
touch lambda/function_before.py lambda/function_after.py && \
rm requirements.txt requirements-dev.txt main.py

主なプロジェクト構成は以下のようになります:

jmespath/
├── jmespath/
│   └── jmespath_stack.py     # AWS CDK スタック定義
├── lambda/                   # Lambda 関数のソースコード
│   ├── function_after.py     # JMESPath Functions 機能使用版
│   └── function_before.py    # JMESPath Functions 機能未使用版
├── app.py                    # AWS CDK アプリケーションのエントリーポイント
├── cdk.json                  # AWS CDK 設定ファイル
├── pyproject.toml            # 依存関係管理ファイル
└── uv.lock                   # 依存関係管理ファイル

1. 依存関係の設定

Terminal

# Powertools for AWS Lambda, jmespath の追加
uv add --group lambda aws-lambda-powertools

2. Lambda 関数の実装 (JMESPath Functions 機能未使用版)

lambda/function_before.py

import json
import base64
from datetime import datetime, timezone
from typing import Any

from aws_lambda_powertools.utilities.typing import LambdaContext


def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
    """JMESPath Functions 機能未使用"""

    try:
        # リクエストボディの取得・デコード
        body: str = event.get("body", "{}")
        if event.get("isBase64Encoded", False):
            body = base64.b64decode(body).decode('utf-8')

        request_data = json.loads(body)

        # ユーザー情報の抽出
        users: list = request_data.get("data", {}).get("users", [])

        # 条件に基づくフィルタリング
        active_users: list = []
        for user in users:
            if user.get("status") == "active" and user.get("age", 0) >= 18:
                active_users.append({
                    "id": user.get("id"),
                    "name": user.get("profile", {}).get("name"),
                    "email": user.get("contact", {}).get("email"),
                    "department": user.get("work", {}).get("department")
                })

        # レスポンスの構築
        response_data: dict = {
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "total_users": len(users),
            "active_users_count": len(active_users),
            "active_users": active_users,
            "source_ip": event.get("requestContext", {}).get("identity", {}).get("sourceIp")
        }

        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json"
            },
            "body": json.dumps(response_data)
        }

    except Exception as e:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": str(e)})
        }

3. Lambda 関数の実装 (JMESPath Functions 機能使用版)

lambda/function_after.py

import json
from datetime import datetime, timezone
from typing import Any

from aws_lambda_powertools.utilities.typing import LambdaContext
from aws_lambda_powertools.utilities.jmespath_utils import query


def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
    """JMESPath Functions 機能使用"""

    try:
        # リクエストボディの取得・デコード
        body_query: str = """
        isBase64Encoded && powertools_json(powertools_base64(body)) || powertools_json(body)
        """
        request_data = query(data=event, envelope=body_query)

        # 条件に基づくフィルタリング
        active_users_query: str = """
        data.users[?status == 'active' && age >= `18`].{
            id: id,
            name: profile.name,
            email: contact.email,
            department: work.department
        }
        """
        active_users = query(data=request_data, envelope=active_users_query)

        # レスポンスの構築
        response_data: dict = {
            "processed_at": datetime.now(timezone.utc).isoformat(),
            "total_users": query(data=request_data, envelope="length(data.users)"),
            "active_users_count": len(active_users) if active_users else 0,
            "active_users": active_users or [],
            "source_ip": query(data=event, envelope="requestContext.identity.sourceIp")
        }

        return {
            "statusCode": 200,
            "headers": {
                "Content-Type": "application/json"
            },
            "body": json.dumps(response_data)
        }

    except Exception as e:
        return {
            "statusCode": 400,
            "body": json.dumps({"error": str(e)})
        }

4. AWS CDK スタックの定義

jmespath/jmespath_stack.py

from aws_cdk import CfnOutput, Stack
from aws_cdk import aws_lambda as _lambda
from constructs import Construct


class JmespathStack(Stack):
    def __init__(self, scope: Construct, construct_id: str, **kwargs) -> None:
        super().__init__(scope, construct_id, **kwargs)

        # Lambda Layer: Powertools for AWS Lambda (Python)
        # Layer ARN: https://docs.powertools.aws.dev/lambda/python/latest/#lambda-layer
        layer = _lambda.LayerVersion.from_layer_version_arn(
            self,
            "PowertoolsLayer",
            layer_version_arn="arn:aws:lambda:ap-northeast-1:017000801446:layer:AWSLambdaPowertoolsPythonV3-python311-x86_64:20"
        )

        # Lambda 関数: JMESPath Functions 機能未使用版
        function_before = _lambda.Function(
            self,
            "FunctionBefore",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="function_before.lambda_handler",
            code=_lambda.Code.from_asset("lambda"),
            layers=[layer],
        )

        # Lambda 関数: JMESPath Functions 機能使用版
        function_after = _lambda.Function(
            self,
            "FunctionAfter",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="function_after.lambda_handler",
            code=_lambda.Code.from_asset("lambda"),
            layers=[layer],
        )

        CfnOutput(
            self,
            "FunctionBeforeName",
            value=function_before.function_name,
            export_name="FunctionBeforeName",
        )

        CfnOutput(
            self,
            "FunctionAfterName",
            value=function_after.function_name,
            export_name="FunctionAfterName",
        )

5. デプロイ・動作確認

Terminal

# デプロイの実行
cdk deploy --require-approval never

Terminal

# テスト用のペイロード作成
cat > test_payload.json << 'EOF'
{
  "body": "eyJkYXRhIjp7InVzZXJzIjpbeyJpZCI6MSwic3RhdHVzIjoiYWN0aXZlIiwiYWdlIjoyNSwicHJvZmlsZSI6eyJuYW1lIjoiQWxpY2UifSwiY29udGFjdCI6eyJlbWFpbCI6ImFsaWNlQGV4YW1wbGUuY29tIn0sIndvcmsiOnsiZGVwYXJ0bWVudCI6IkVuZ2luZWVyaW5nIn19LHsiaWQiOjIsInN0YXR1cyI6ImluYWN0aXZlIiwiYWdlIjozMCwicHJvZmlsZSI6eyJuYW1lIjoiQm9iIn0sImNvbnRhY3QiOnsiZW1haWwiOiJib2JAZXhhbXBsZS5jb20ifSwid29yayI6eyJkZXBhcnRtZW50IjoiU2FsZXMifX0seyJpZCI6Mywic3RhdHVzIjoiYWN0aXZlIiwiYWdlIjoxNywicHJvZmlsZSI6eyJuYW1lIjoiQ2hhcmxpZSJ9LCJjb250YWN0Ijp7ImVtYWlsIjoiY2hhcmxpZUBleGFtcGxlLmNvbSJ9LCJ3b3JrIjp7ImRlcGFydG1lbnQiOiJNYXJrZXRpbmcifX1dfX0=",
  "isBase64Encoded": true,
  "requestContext": {
    "identity": {
      "sourceIp": "192.168.1.1"
    }
  }
}
EOF

Terminal

# JMESPath Functions 機能未使用版のテスト
aws lambda invoke \
  --function-name <Before 関数名> \
  --cli-binary-format raw-in-base64-out \
  --payload file://test_payload.json \
  response_before.json && \
cat response_before.json | \
jq -r ".body | fromjson | .active_users"

Terminal

# JMESPath Functions 機能使用版のテスト
aws lambda invoke \
  --function-name <After 関数名> \
  --cli-binary-format raw-in-base64-out \
  --payload file://test_payload.json \
  response_after.json && \
cat response_after.json | \
jq -r ".body | fromjson | .active_users"

6. リソースのクリーンアップ

Terminal

cdk destroy -f

JMESPath Functions 機能を使用するメリット

保守性の向上

• 可読性: データ処理ロジックが直感的, 一行のクエリで複雑なデータ抽出が可能
• エンコーディング: Base64やGzip処理の簡略化
• JSON処理: 文字列とオブジェクト間の変換
• 再利用性: 共通的なクエリパターンの標準化

まとめ

本記事では、Powertools for AWS Lambda (Python) の JMESPath Functions 機能について紹介しました。
特に JSON データを多く扱う Lambda 関数において、この機能はとても有用だと思います。

Powertools for AWS Lambda に興味のある方、Lambda 関数での開発方法に課題を感じている方の参考になれば嬉しいです。

今後も Powertools for AWS Lambda の各種機能についてブログを書いていく予定ですので、お楽しみに。

参考資料

• Powertools for AWS Lambda (Python) 公式ドキュメント
• JMESPath Functions 機能ドキュメント
• Powertools for AWS 公式 GitHub Repository
• JMESPath 公式サイト

最後に

株式会社 Penetrator はシリーズ A ラウンドにおいて総額 5.5 億円の資金調達を実施し、不動産テック業界における更なる成長を目指して、採用活動を一層強化しています。
エンジニア, デザイナー, カスタマーサクセス, BizDev, 営業, マーケティングなど、事業拡大を支える多様なポジションで共に挑戦していただける方を待っています!!

▽ 会社のカルチャーを知りたい方はこちら

https://www.wantedly.com/companies/company_9924832

▽ 募集職種を知りたい方はこちら

https://hrmos.co/pages/where/jobs