これは、なにをしたくて書いたもの?
AWS Lambda関数をTypeScriptで書く場合には、型定義を@types/aws-lambdaから使うことが多いと思います。
では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]:
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_bodyでJSONパース後のオブジェクトが
取得できるようです。
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 GatewayのREST 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関数の引数には適用できなかったのがちょっと残念でしたが、使い方はわかったのでよしとしましょう。