ステップ 2: Lambda 関数を作成する

Lambda 関数を作成するには、以下の手順を実行します。

ステップ 2.1: Lambda 関数を作成する

このチュートリアルでは、Lambda 関数のコード例を Python で提供します。

Python

次の Python コードの例では、 ElastiCache 項目をキャッシュに読み書きします。コードを app.py という名前のファイルに保存します。コードのelasticache_endpoint値を、ステップ 1.2 でコピーしたエンドポイントアドレスに置き換えてください。

from typing import Tuple, Union
from urllib.parse import ParseResult, urlencode, urlunparse

import botocore.session
import redis
from botocore.model import ServiceId
from botocore.signers import RequestSigner
from cachetools import TTLCache, cached
import uuid

class ElastiCacheIAMProvider(redis.CredentialProvider):
    def __init__(self, user, cache_name, is_serverless=False, region="us-east-1"):
        self.user = user
        self.cache_name = cache_name
        self.is_serverless = is_serverless
        self.region = region

        session = botocore.session.get_session()
        self.request_signer = RequestSigner(
            ServiceId("elasticache"),
            self.region,
            "elasticache",
            "v4",
            session.get_credentials(),
            session.get_component("event_emitter"),
        )

    # Generated IAM tokens are valid for 15 minutes
    @cached(cache=TTLCache(maxsize=128, ttl=900))
    def get_credentials(self) -> Union[Tuple[str], Tuple[str, str]]:
        query_params = {"Action": "connect", "User": self.user}
        if self.is_serverless:
            query_params["ResourceType"] = "ServerlessCache"
        url = urlunparse(
            ParseResult(
                scheme="https",
                netloc=self.cache_name,
                path="/",
                query=urlencode(query_params),
                params="",
                fragment="",
            )
        )
        signed_url = self.request_signer.generate_presigned_url(
            {"method": "GET", "url": url, "body": {}, "headers": {}, "context": {}},
            operation_name="connect",
            expires_in=900,
            region_name=self.region,
        )
        # RequestSigner only seems to work if the URL has a protocol, but
        # Elasticache only accepts the URL without a protocol
        # So strip it off the signed URL before returning
        return (self.user, signed_url.removeprefix("https://"))

def lambda_handler(event, context):
    username = "iam-user-01" # replace with your user id
    cache_name = "cache-01" # replace with your cache name
    elasticache_endpoint = "cache-01-xxxxx.serverless.use1.cache.amazonaws.com" # replace with your cache endpoint
    creds_provider = ElastiCacheIAMProvider(user=username, cache_name=cache_name, is_serverless=True)
    redis_client = redis.Redis(host=elasticache_endpoint, port=6379, credential_provider=creds_provider, ssl=True, ssl_cert_reqs="none")
    
    key='uuid'
    # create a random UUID - this will be the sample element we add to the cache
    uuid_in = uuid.uuid4().hex
    redis_client.set(key, uuid_in)
    result = redis_client.get(key)
    decoded_result = result.decode("utf-8")
    # check the retrieved item matches the item added to the cache and print
    # the results
    if decoded_result == uuid_in:
        print(f"Success: Inserted {uuid_in}. Fetched {decoded_result} from Redis OSS.")
    else:
        raise Exception(f"Bad value retrieved. Expected {uuid_in}, got {decoded_result}")
        
    return "Fetched value from Redis"

このコードでは、Python redis-py ライブラリを使用して項目をキャッシュに入れ、取得します。このコードは cachetools を使用して、生成された IAM 認証トークンを 15 分間キャッシュします。redis-py と cachetools を含むデプロイパッケージを作成するには、次の手順を実行します。

app.py ソースコードファイルを含むプロジェクトディレクトリで、フォルダパッケージを作成して redis-py ライブラリと cachetools ライブラリをインストールします。

mkdir package

pip を使用して redis-py、cachetools をインストールします。

pip install --target ./package redis
pip install --target ./package cachetools

redis-py ライブラリと cachetools ライブラリを含む .zip ファイルを作成します。Linux および macOS では、次のコマンドを実行します。Windows では、任意の zip ユーティリティを使用して、ルートに redis-py ライブラリと cachetools ライブラリを含む .zip ファイルを作成します。

cd package
zip -r ../my_deployment_package.zip .

.zip ファイルに関数コードを追加します。Linux および macOS では、次のコマンドを実行します。Windows では、任意の zip ユーティリティを使用して、app.py を .zip ファイルのルートに追加します。

cd ..
zip my_deployment_package.zip app.py

ステップ 2.2: IAM ロール (実行ロール) を作成する

という名前の AWS マネージドポリシーをロールAWSLambdaVPCAccessExecutionRoleにアタッチします。

aws iam attach-role-policy \
 --role-name "elasticache-iam-auth-app" \
 --policy-arn "arn:aws:iam::aws:policy/service-role/AWSLambdaVPCAccessExecutionRole"

ステップ 2.3: デプロイパッケージをアップロードする (Lambda 関数を作成する)

このステップでは、create-function AWS CLI コマンドを使用して Lambda 関数 (AccessRedis) を作成します。

デプロイパッケージ .zip ファイルを含むプロジェクトディレクトリから、次の Lambda CLI create-function コマンドを実行します。

ロールオプションには、ステップ 2.2 で作成した実行ロールの ARN を使用します。vpc-config には、デフォルトの VPC のサブネットとデフォルトの VPC のセキュリティグループ ID のカンマ区切りリストを入力します。これらの値は Amazon VPC コンソール にあります。デフォルトの VPC のサブネットを検索するには、VPCs を選択し、 AWS アカウントのデフォルト VPC を選択します。この VPC のセキュリティグループを検索するには、「セキュリティ」に移動し、「セキュリティグループ」を選択します。us-east-1 リージョンが選択されていることを確認します。

aws lambda create-function \
--function-name AccessRedis  \
--region us-east-1 \
--zip-file fileb://my_deployment_package.zip \
--role arn:aws:iam::123456789012:role/elasticache-iam-auth-app \
--handler app.lambda_handler \
--runtime python3.12  \
--timeout 30 \
--vpc-config SubnetIds=comma-separated-vpc-subnet-ids,SecurityGroupIds=default-security-group-id