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

はじめに

こんにちは。
株式会社 WHERE のプロダクト開発部で主にインフラを担当している えんどぅ です。

みなさん元気に Lambda Life を送っていますか??

Lambda 関数の開発において、設定値や機密情報の管理は基本的には避けて通れない課題だと思います。
API キー、外部サービスのエンドポイント URL など、環境ごとに異なる設定を安全かつ効率的に管理する必要があります。

しかし、これらの値を直接コードに埋め込んだり、環境変数のみで管理する場合、セキュリティやパフォーマンス、デプロイ頻度等の課題に直面することがあると思います。

そんな課題を解決してくれるのが、Powertools for AWS Lambda の Parameters 機能です。

今回はこの機能について、実際の使用イメージと共に紹介します。

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

※ 過去のブログ記事はコチラ👇

• Typing 編
• JMESPath 編
• Validation 編
• Parser 編

Powertools for AWS Lambda とは

Powertools for AWS Lambda は、サーバーレスベストプラクティスの実装と開発者の生産性向上を支援する開発者ツールキットです。

段階的に導入できる柔軟な設計により、必要な機能から順次採用することができます。

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

Parameters 機能とは

Parameters 機能 は、AWS Systems Manager Parameter Store, AWS Secrets Manager, AWS AppConfig, Amazon DynamoDB などのパラメータストアから、パラメータ値を簡単に取得・キャッシュできるユーティリティです。

主な特徴

複数のパラメータストアへの統一的なアクセス

• SSM Parameter Store, Secrets Manager, AppConfig にデフォルトで対応
• 統一された API で異なるストアからパラメータを取得

自動キャッシュによるパフォーマンス最適化

• デフォルトで 5 分間のインメモリキャッシュ
• API コール数の削減とレスポンスタイムの改善
• キャッシュ時間のカスタマイズが可能

データ変換の自動化

• JSON や Base64 エンコードされたデータの自動デコード
• カスタム変換処理の実装サポート

柔軟な取得方法

• 単一パラメータの取得
• パス配下の複数パラメータの一括取得
• 名前指定での複数パラメータ取得

実践: 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

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

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

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

# プロジェクトのセットアップ
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.py && \
rm requirements.txt requirements-dev.txt main.py

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

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

1. 依存関係の設定

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

2. Lambda 関数の実装

lambda/function.py

import json
from typing import Any

from aws_lambda_powertools.utilities import parameters
from aws_lambda_powertools.utilities.parameters.ssm import get_parameters_by_name
from aws_lambda_powertools.utilities.typing import LambdaContext


def lambda_handler(event: dict[str, Any], context: LambdaContext) -> dict[str, Any]:
    """Parameters 機能を使用した実装"""

    try:
        # === 基本的な使い方 ===

        # SSM Parameter Store から設定値を取得（自動キャッシュ）
        api_endpoint = parameters.get_parameter("/myapp/api/endpoint")

        # JSON 形式のパラメータを自動的にパース
        db_config = parameters.get_parameter("/myapp/database/config", transform="json")

        # Secrets Manager から機密情報を取得（自動キャッシュ）
        api_key = parameters.get_secret("/myapp/api/key")

        # 暗号化されたパラメータを自動復号化
        # NOTE: CDK で Secure String 型の parameter を定義できないため今回は省略
        # encrypted_value = parameters.get_parameter(
        #     "/myapp/encrypted/value", decrypt=True
        # )

        # === 応用: 複数パラメータの一括取得 ===

        # パス配下の全パラメータを一括取得
        all_configs = parameters.get_parameters(
            "/myapp/config",
            max_age=300,  # 5分間キャッシュ
        )

        # 特定の名前のパラメータを個別設定で取得
        specific_params = get_parameters_by_name(
            parameters={
                "/myapp/database/config": {
                    "transform": "json",
                    "max_age": 600,  # 10分間キャッシュ
                },
                "/myapp/feature/flags": {
                    "transform": "json",
                    "max_age": 0,  # キャッシュなし
                },
            },
            raise_on_error=False,  # 例外をスローするかどうか
        )

        # ビジネスロジックの実行
        result = process_request(
            api_endpoint=api_endpoint,
            db_config=db_config,
            api_key=api_key,
            all_configs=all_configs,
            specific_params=specific_params,
        )

        return {"statusCode": 200, "body": json.dumps(result, default=str)}

    except parameters.exceptions.GetParameterError as e:
        return {"statusCode": 500, "body": json.dumps({"error": str(e)})}


def process_request(
    api_endpoint: str,
    db_config: dict,
    api_key: str,
    all_configs: dict,
    specific_params: dict,
) -> dict[str, Any]:
    """ビジネスロジック処理"""
    return {
        "message": "Request processed successfully",
        "endpoint": api_endpoint,
        "db_host": db_config.get("host"),
        "has_api_key": bool(api_key),
        "config_count": len(all_configs),
        "specific_params_count": len(specific_params),
    }

ポイント:

• parameters.get_parameter() でシンプルにパラメータ取得
• transform="json" で JSON の自動パース
• decrypt=True で暗号化パラメータの自動復号化
• デフォルトで 5 分間のキャッシュが有効
• get_parameters() でパス配下の一括取得
• get_parameters_by_name() でパラメータごとに異なる設定を適用

3. AWS CDK スタックの定義

parameters/parameters_stack.py

import json

from aws_cdk import CfnOutput, Duration, SecretValue, Stack
from aws_cdk import aws_iam as iam
from aws_cdk import aws_lambda as _lambda
from aws_cdk import aws_secretsmanager as secretsmanager
from aws_cdk import aws_ssm as ssm
from constructs import Construct


class ParametersStack(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 = _lambda.LayerVersion.from_layer_version_arn(
            self,
            "PowertoolsLayer",
            layer_version_arn="arn:aws:lambda:us-east-1:017000801446:layer:AWSLambdaPowertoolsPythonV3-python313-x86_64:23",
        )

        # SSM Parameter Store パラメータの作成
        api_endpoint_param = ssm.StringParameter(
            self,
            "ApiEndpoint",
            parameter_name="/myapp/api/endpoint",
            string_value="https://api.example.com/v1",
            description="API endpoint URL",
        )

        db_config_param = ssm.StringParameter(
            self,
            "DatabaseConfig",
            parameter_name="/myapp/database/config",
            string_value=json.dumps(
                {"host": "db.example.com", "port": 5432, "database": "myapp"}
            ),
            description="Database configuration (JSON)",
        )

        config_timeout = ssm.StringParameter(
            self,
            "ConfigTimeout",
            parameter_name="/myapp/config/timeout",
            string_value="30",
            description="Request timeout in seconds",
        )

        config_retry = ssm.StringParameter(
            self,
            "ConfigRetry",
            parameter_name="/myapp/config/retry",
            string_value="3",
            description="Number of retries",
        )

        config_log_level = ssm.StringParameter(
            self,
            "ConfigLogLevel",
            parameter_name="/myapp/config/log_level",
            string_value="INFO",
            description="Logging level",
        )

        feature_flags_param = ssm.StringParameter(
            self,
            "FeatureFlags",
            parameter_name="/myapp/feature/flags",
            string_value=json.dumps(
                {"new_feature_enabled": True, "beta_features": False}
            ),
            description="Feature flags configuration (JSON)",
        )

        # Secrets Manager シークレットの作成
        api_key_secret = secretsmanager.Secret(
            self,
            "ApiKeySecret",
            secret_name="/myapp/api/key",
            description="API Key for external service",
            secret_string_value=SecretValue.unsafe_plain_text("my-secret-key"),
        )

        # Lambda 関数
        function = _lambda.Function(
            self,
            "ParametersFunction",
            runtime=_lambda.Runtime.PYTHON_3_11,
            handler="function.lambda_handler",
            code=_lambda.Code.from_asset("lambda"),
            layers=[layer],
            timeout=Duration.seconds(30),
            environment={
                "POWERTOOLS_SERVICE_NAME": "myapp",
                "POWERTOOLS_PARAMETERS_MAX_AGE": "300",  # デフォルトキャッシュ時間: 5分
            },
        )

        # IAM ポリシーの付与
        api_endpoint_param.grant_read(function)
        db_config_param.grant_read(function)
        feature_flags_param.grant_read(function)
        api_key_secret.grant_read(function)

        function.add_to_role_policy(
            statement=iam.PolicyStatement(
                actions=["ssm:GetParametersByPath"],
                resources=[
                    f"arn:aws:ssm:{self.region}:{self.account}:parameter/myapp/config"
                ],
            )
        )

        CfnOutput(
            self,
            "FunctionName",
            value=function.function_name,
            export_name="ParametersFunctionName",
        )

4. デプロイ・動作確認

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

# テスト用のペイロード作成
cat > test_payload.json << 'EOF'
{
  "action": "test"
}
EOF

# Lambda 関数のテスト（初回実行）
aws lambda invoke \
  --function-name <関数名> \
  --cli-binary-format raw-in-base64-out \
  --payload file://test_payload.json \
  response.json && \
cat response.json | jq

# 2回目の実行でキャッシュ効果を確認
aws lambda invoke \
  --function-name <関数名> \
  --cli-binary-format raw-in-base64-out \
  --payload file://test_payload.json \
  response_cached.json && \
cat response_cached.json | jq

5. パフォーマンスの確認

CloudWatch Logs で実行時間を確認すると、2 回目以降の実行でキャッシュが効いていることが確認できます:

初回実行（キャッシュなし）:

• Duration: 約 4422ms

2回目以降の実行（キャッシュあり）:

• Duration: 約 240ms

キャッシュにより、実行時間が大幅に短縮される事が確認できます。

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

cdk destroy -f

Parameters 機能を使用するメリット

コードの簡潔性と可読性の向上

• Parameters 機能を使用することで、boto3 を直接使う場合と比べて大幅にコードが簡潔になります。

自動キャッシュによるパフォーマンス向上

• API コール削減
• コスト削減:

柔軟なキャッシュ制御

# パラメータごとに異なるキャッシュ時間を設定
config = parameters.get_parameter("/config", max_age=600)  # 10分
feature_flags = parameters.get_parameter("/flags", max_age=60)  # 1分
no_cache = parameters.get_parameter("/realtime", max_age=0)  # キャッシュなし

# 強制的に最新値を取得
latest_value = parameters.get_parameter("/config", force_fetch=True)

データ変換の自動化

# JSON 形式のパラメータを自動パース
db_config = parameters.get_parameter("/db/config", transform="json")

# Base64 エンコードされたデータを自動デコード
binary_data = parameters.get_parameter("/binary", transform="binary")

# サフィックスから自動判定
# /config.json → JSON としてパース
# /data.binary → Base64 デコード
auto_transformed = parameters.get_parameter("/config.json", transform="auto")

まとめ

本記事では、Powertools for AWS Lambda (Python) の Parameters 機能の基本的な部分について主に紹介しました。

自動キャッシュによるパフォーマンス向上、複数のパラメータストアへの統一的なアクセス、データ変換の自動化など、Lambda 関数開発において非常に有用だと思います。

Powertools for AWS Lambda に興味のある方、Lambda 関数でのパラメータ管理に課題を感じている方の参考になれば嬉しいです。

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

参考資料

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

最後に

株式会社 WHERE (旧: 株式会社 Penetrator) は、シリーズ A ラウンドにおいて総額 5.5 億円の資金調達を実施し、不動産テック業界における更なる成長を目指して採用活動を一層強化しています。

エンジニア, デザイナー, カスタマーサクセス, BizDev, 営業, マーケティングなど、事業拡大を支える多様なポジションで共に挑戦していただける方を待っています!!

▽ 会社のカルチャーを知りたい方はこちら
https://www.wantedly.com/companies/company_9924832

▽ 募集職種を知りたい方はこちら
https://www.wantedly.com/companies/company_9924832/projects