CLOVER🍀

That was when it all began.

Powertools for AWS Lambda(Python)を使って、AWS Lambda関数を書く時に型定義を利用する

これは、なにをしたくて書いたもの?

AWS Lambda関数をTypeScriptで書く場合には、型定義を@types/aws-lambdaから使うことが多いと思います。

@types/aws-lambda - npm

ではPythonの場合はどうしたらいいのだろうと調べてみたら、Powertools for AWS Lambda(Python)に含まれているようだったので
こちらを試してみることにしました。

Powertools for AWS Lambda(Python)のTypingとEvent Source Data Classesを使う

Powertools for AWS Lambda(Python)のTypingでは、AWS LambdaのContextに関する型を提供してくれるようです。

Typing - Powertools for AWS Lambda (Python)

ではAWS Lambda関数が受け取るイベントに関する型は?というと、Event Source Data Classesを使うようです。

Event Source Data Classes - Powertools for AWS Lambda (Python)

Event Source Data Classesではイベントの型定義、ネストされたフィールドのデコード/デシリアライズを行うためのヘルパー関数を
提供してくれるようです。

なお、Powertools for AWS Lambda(Python)のAPIリファレンスはこちら。

aws_lambda_powertools API documentation

今回使うのはこのあたりですね。

aws_lambda_powertools.utilities.typing API documentation

aws_lambda_powertools.utilities.data_classes API documentation

では、試してみましょう。

環境

今回の環境はこちら。

$ python3 --version
Python 3.12.3


$ pip3 --version
pip 24.0 from /usr/lib/python3/dist-packages/pip (python 3.12)

動作確認はLocalStackで行うことにします。

$ localstack --version
LocalStack CLI 4.0.3

起動。

$ localstack start

AWS CLIおよびAWS SAMのLocalStack版。

$ awslocal --version
aws-cli/2.22.17 Python/3.12.6 Linux/6.8.0-50-generic exe/x86_64.ubuntu.24


$ samlocal --version
SAM CLI, version 1.132.0

AWS SAMプロジェクトを作成して、Powertools for AWS Lambda(Python)を導入する

それでは、まずはAWS SAMプロジェクトを作成します。

$ samlocal init --name hello-powertools-typing --runtime python3.12 --app-template hello-world --package-type Zip --no-tracing --no-application-insights --structured-logging
$ cd hello-powertools-typing

生成されたAWS Lambda関数の定義はこちら。

hello_world/app.py

import json

# import requests


def lambda_handler(event, context):
    """Sample pure Lambda function

    Parameters
    ----------
    event: dict, required
        API Gateway Lambda Proxy Input Format

        Event doc: https://docs.aws.amazon.com/apigateway/latest/developerguide/set-up-lambda-proxy-integrations.html#api-gateway-simple-proxy-for-lambda-input-format

    context: object, required
        Lambda Context runtime methods and attributes

        Context doc: https://docs.aws.amazon.com/lambda/latest/dg/python-context-object.html

    Returns
    ------
    API Gateway Lambda Proxy Output Format: dict

        Return doc: https://docs.aws.amazon.com/apigateway/latest/developerguide/set-up-lambda-proxy-integrations.html
    """

    # try:
    #     ip = requests.get("http://checkip.amazonaws.com/")
    # except requests.RequestException as e:
    #     # Send some context about this error to Lambda Logs
    #     print(e)

    #     raise e

    return {
        "statusCode": 200,
        "body": json.dumps({
            "message": "hello world",
            # "location": ip.text.replace("\n", "")
        }),
    }

こちらを変更して、型も導入するようにしましょう。

AWS Lambda関数のコードが配置してあるディレクトリーに移動。

$ cd hello_world

Powertools for AWS Lambda(Python)とmypyをインストール。

$ pip3 install aws-lambda-powertools mypy

インストールされた依存関係の一覧。

$ pip3 list
Package               Version
--------------------- -------
aws-lambda-powertools 3.3.0
jmespath              1.0.1
mypy                  1.13.0
mypy-extensions       1.0.0
pip                   24.0
typing_extensions     4.12.2

requirements.txtも作成しておきましょう。

$ pip3 freeze > requirements.txt

app.pyはこのように変更しました。

app.py

from aws_lambda_powertools.utilities.data_classes import APIGatewayProxyEvent
from aws_lambda_powertools.utilities.typing import LambdaContext

import json
import logging

logger = logging.getLogger()
logger.setLevel("INFO")

def lambda_handler(event: dict, context: LambdaContext) -> dict[str, object]:
    api_gateway_proxy_event = APIGatewayProxyEvent(event)

    logger.info(f"api_gateway_proxy_event = {api_gateway_proxy_event}")
    logger.info(f"remaining_time_in_millis = {context.get_remaining_time_in_millis()}")

    json_body = api_gateway_proxy_event.json_body

    return {
        "statusCode": 200,
        "body": json.dumps({
            "message": json_body["message"],
        }),
    }

LambdaContextはPowertools for AWS Lambda(Python)による型ですね。

def lambda_handler(event: dict, context: LambdaContext) -> dict[str, object]:

Typing / LambdaContext

APIGatewayProxyEventは受け取ったイベントから変換しています。

    api_gateway_proxy_event = APIGatewayProxyEvent(event)

Event Source Data Classes / Getting started

Event Source Data Classes / Supported event sources / API Gateway Proxy

以下のように@event_sourceデコレーターを使う方法もあるようなのですが、LocalStackでは動きませんでした…。
AWS Lambda関数が起動しなくなりました

@event_source(data_classes=APIGatewayProxyEvent)
def lambda_handler(event: APIGatewayProxyEvent, context: LambdaContext):

なお、Content-Typeがapplication/jsonなら以下のようにAPIGatewayProxyEvent#json_bodyJSONパース後のオブジェクトが
取得できるようです。

    json_body = api_gateway_proxy_event.json_body

AWS SAMテンプレートの方は、AWS Lambda関数がHTTPボディを受け取るようにしているのでHTTPメソッドをPOSTに変更しています。

Resources:
  HelloWorldFunction:
    Type: AWS::Serverless::Function # More info about Function Resource: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#awsserverlessfunction
    Properties:
      CodeUri: hello_world/
      Handler: app.lambda_handler
      Runtime: python3.12
      Architectures:
      - x86_64
      Events:
        HelloWorld:
          Type: Api # More info about API Event Source: https://github.com/awslabs/serverless-application-model/blob/master/versions/2016-10-31.md#api
          Properties:
            Path: /hello
            Method: post

ビルドしてデプロイ。

$ samlocal build --no-cached && samlocal deploy --region us-east-1

リソースが作成されました。

CloudFormation events from stack operations (refresh every 5.0 seconds)
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
ResourceStatus                             ResourceType                               LogicalResourceId                          ResourceStatusReason
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
CREATE_IN_PROGRESS                         AWS::CloudFormation::Stack                 hello-powertools-typing                    -
CREATE_IN_PROGRESS                         AWS::IAM::Role                             HelloWorldFunctionRole                     -
CREATE_COMPLETE                            AWS::IAM::Role                             HelloWorldFunctionRole                     -
CREATE_IN_PROGRESS                         AWS::Lambda::Function                      HelloWorldFunction                         -
CREATE_COMPLETE                            AWS::Lambda::Function                      HelloWorldFunction                         -
CREATE_IN_PROGRESS                         AWS::ApiGateway::RestApi                   ServerlessRestApi                          -
CREATE_COMPLETE                            AWS::ApiGateway::RestApi                   ServerlessRestApi                          -
CREATE_IN_PROGRESS                         AWS::Lambda::Permission                    HelloWorldFunctionHelloWorldPermissionPr   -
                                                                                      od
CREATE_COMPLETE                            AWS::Lambda::Permission                    HelloWorldFunctionHelloWorldPermissionPr   -
                                                                                      od
CREATE_IN_PROGRESS                         AWS::ApiGateway::Deployment                ServerlessRestApiDeploymentd4d193690c      -
CREATE_COMPLETE                            AWS::ApiGateway::Deployment                ServerlessRestApiDeploymentd4d193690c      -
CREATE_IN_PROGRESS                         AWS::ApiGateway::Stage                     ServerlessRestApiProdStage                 -
CREATE_COMPLETE                            AWS::ApiGateway::Stage                     ServerlessRestApiProdStage                 -
CREATE_COMPLETE                            AWS::CloudFormation::Stack                 hello-powertools-typing                    -
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------

CloudFormation outputs from deployed stack
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Outputs
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Key                 HelloWorldApi
Description         API Gateway endpoint URL for Prod stage for Hello World function
Value               https://orgd7p7vlo.execute-api.amazonaws.com:4566/Prod/hello/

Key                 HelloWorldFunction
Description         Hello World Lambda Function ARN
Value               arn:aws:lambda:us-east-1:000000000000:function:hello-powertools-typing-HelloWorldFunction-01f8122a

Key                 HelloWorldFunctionIamRole
Description         Implicit IAM Role created for Hello World function
Value               arn:aws:iam::000000000000:role/hello-powertools-typing-HelloWorldFunctionRole-2566e5b4
-------------------------------------------------------------------------------------------------------------------------------------------------------------------------


Successfully created/updated stack - hello-powertools-typing in us-east-1

Amazon API GatewayREST APIのIDを取得。

$ REST_API_ID=$(awslocal apigateway get-rest-apis --query 'reverse(sort_by(items[], &createdDate))[0].id' --output text)

確認。

$ curl -XPOST -H 'Content-Type: application/json' localhost:4566/_aws/execute-api/$REST_API_ID/Prod/hello -d '{"message": "Hello, Powertools for AWS Lambda!!"}'
{"message": "Hello, Powertools for AWS Lambda!!"}(venv) 

OKですね。

ログも見ておきます。

$ awslocal logs tail --follow /aws/lambda/hello-powertools-typing-HelloWorldFunction-01f8122a
2024-12-14T12:23:02.895000+00:00 2024/12/14/[$LATEST]1fda1a159d25fda07adda23c315ea704 START RequestId: e5c52613-82f4-499c-b319-f3624473402f Version: $LATEST
2024-12-14T12:23:02.948000+00:00 2024/12/14/[$LATEST]1fda1a159d25fda07adda23c315ea704 [INFO]    2024-12-14T12:23:02.887Z        e5c52613-82f4-499c-b319-f3624473402f    api_gateway_proxy_event = {'body': '{"message": "Hello, Powertools for AWS Lambda!!"}', 'headers': {'host': 'localhost:4566', 'user-agent': 'curl/8.5.0', 'accept': '*/*', 'content-type': 'application/json', 'x-forwarded-for': '172.17.0.1', 'x-forwarded-port': '4566', 'x-forwarded-proto': 'HTTP', 'x-amzn-trace-id': 'Root=1-675d78a2-d51cebd94b69f262bb5c61a9;Parent=f744169f8d4752d9;Sampled=0'}, 'http_method': 'POST', 'is_base64_encoded': False, 'multi_value_headers': {'host': ['localhost:4566'], 'user-agent': ['curl/8.5.0'], 'accept': ['*/*'], 'content-type': ['application/json'], 'x-forwarded-for': ['172.17.0.1'], 'x-forwarded-port': ['4566'], 'x-forwarded-proto': ['HTTP'], 'x-amzn-trace-id': ['Root=1-675d78a2-d51cebd94b69f262bb5c61a9;Parent=f744169f8d4752d9;Sampled=0']}, 'multi_value_query_string_parameters': {}, 'path': '/hello', 'path_parameters': {}, 'query_string_parameters': {}, 'raw_event': '[SENSITIVE]', 'request_context': {'account_id': '000000000000', 'api_id': 'orgd7p7vlo', 'authorizer': {'claims': {}, 'integration_latency': None, 'principal_id': '', 'raw_event': '[SENSITIVE]', 'scopes': []}, 'connected_at': None, 'connection_id': None, 'domain_name': 'localhost:4566', 'domain_prefix': 'localhost:4566', 'event_type': None, 'extended_request_id': '3e0584b8', 'http_method': 'POST', 'identity': {'access_key': None, 'account_id': None, 'api_key': None, 'api_key_id': None, 'caller': None, 'client_cert': None, 'cognito_authentication_provider': None, 'cognito_authentication_type': None, 'cognito_identity_id': None, 'cognito_identity_pool_id': None, 'principal_org_id': None, 'raw_event': '[SENSITIVE]', 'source_ip': '127.0.0.1', 'user': None, 'user_agent': 'curl/8.5.0', 'user_arn': None}, 'message_direction': None, 'message_id': None, 'operation_name': None, 'path': '/Prod/hello', 'protocol': 'HTTP/1.1', 'raw_event': '[SENSITIVE]', 'request_id': '11ce72a9-33eb-4616-b006-31d482bddd0e', 'request_time': '14/Dec/2024:12:22:58 ', 'request_time_epoch': 1734178978817, 'resource_id': 'zkhsmr96by', 'resource_path': '/hello', 'route_key': None, 'stage': 'Prod'}, 'resolved_headers_field': {'host': ['localhost:4566'], 'user-agent': ['curl/8.5.0'], 'accept': ['*/*'], 'content-type': ['application/json'], 'x-forwarded-for': ['172.17.0.1'], 'x-forwarded-port': ['4566'], 'x-forwarded-proto': ['HTTP'], 'x-amzn-trace-id': ['Root=1-675d78a2-d51cebd94b69f262bb5c61a9;Parent=f744169f8d4752d9;Sampled=0']}, 'resolved_query_string_parameters': {}, 'resource': '/hello', 'stage_variables': {}, 'version': '[Cannot be deserialized]'}
2024-12-14T12:23:03.001000+00:00 2024/12/14/[$LATEST]1fda1a159d25fda07adda23c315ea704 [INFO]    2024-12-14T12:23:02.887Z        e5c52613-82f4-499c-b319-f3624473402f    remaining_time_in_millis = 2997
2024-12-14T12:23:03.054000+00:00 2024/12/14/[$LATEST]1fda1a159d25fda07adda23c315ea704 END RequestId: e5c52613-82f4-499c-b319-f3624473402f
2024-12-14T12:23:03.107000+00:00 2024/12/14/[$LATEST]1fda1a159d25fda07adda23c315ea704 REPORT RequestId: e5c52613-82f4-499c-b319-f3624473402f    Duration: 3.73 ms       Billed Duration: 4 ms       Memory Size: 128 MB     Max Memory Used: 128 MB

よさそうですね。

おわりに

Powertools for AWS Lambda(Python)を使って、AWS Lambda関数を書く時に型定義を利用してみました。

LocalStackではAWS Lambda関数の引数には適用できなかったのがちょっと残念でしたが、使い方はわかったのでよしとしましょう。