はじめに
PythonでLambdaを書くときに自分が意識している内容をアウトプットします。
とりあえずLambdaは使ったことあるよ、という人には1歩進んだネタが提供できるんじゃないかと思います。
開発環境
Lambdaの作成はローカルで行います。
マネジメントコンソール上でもコードを入力して作成できますが、
- Git等のバージョン管理が行われず変更差分を追跡できない
- ユニットテストが行われずソースコードの変更による影響範囲が予測できない
こういった理由で運用時の負債を残してしまう可能性が高いです。
Lambdaと他のAWSリソースとの接続を確認するためにマネジメントコンソール上から動作確認をするケースもあるかと思います。
そういった場合は以降の項で触れるモックを使ったテストを行えばAWS上のリソースを使わなくてもほとんどのAWSリソースとの接続を確認できます。
ローカルでLambdaを作成するにあたって、仮想環境とIaCの設定方法について説明します。
仮想環境（pipenv）
仮想環境はpipenvを使って作成します。
pipenvはインストールしたパッケージとそのバージョンをlockファイルに記録しますが、そのlockファイルを基に同じ環境を再現できる利点があります。
パッケージのインストールにはpipenv installを使用し、開発用のものについては--devオプションを付けます。
本番環境に不要なパッケージを配置しないために、パッケージの扱いを本番用、開発用と明示的に区別します。
pipenv install black --dev # コードフォーマッター / ローカルで使用するため本番環境には不要
pipenv install boto3 --dev # AWS Python SDK / 本番環境では標準で利用できるため不要
pipenv install aws-lambda-powertools --dev # AWS Lambda Powertools Python / 本番環境ではパブリックレイヤーからロードするため不要
pipenv install bs4 # 外部パッケージ / 本番環境での実行に必要なので本番用としてインストール
また、pipenvは.envファイルを作成しておくと仮想環境をアクティベーションした際に環境変数として設定してくれます。
IaC(AWS CDK)
デプロイはIaCで行います。
IaCツールはいくつか種類がありますが、ここではAWS CDKを採用した想定で進めます。
PythonでLambdaを作成する場合、Pythonに特化したaws-lambda-python（CDKv2ではaws-lambda-python-alpha）モジュールを使います。
デプロイ時にrequirements.txtやPipfileに記載したパッケージをCDKが同梱してくれます。
複数の関数で使用するパッケージはレイヤーとして定義し、AWSがパブリックレイヤーとして公開しているものがあればそれを使います。
aws-lambda-powertoolsはパブリックレイヤーの1つです。
CDKコマンドの実行時はenvでパラメーターを分けるために-cオプションを付けます。
-cはコンテキストのオプションで各スタックにパラメーターを渡すことができます。
そのためdevやprdなどのパラメーターでテンプレートを分けられます。
cdk diff -c DEPLOYMENT_ENV={環境名} # コンテキストのKeyは自由に設定可能
cdk deploy -c DEPLOYMENT_ENV={環境名}
cdk destroy -c DEPLOYMENT_ENV={環境名}
CDKでは下記のようにしてコンテキストの値を取得できます。
app = cdk.App()
DEPLOYMENT_ENV = app.node.try_get_context("DEPLOYMENT_ENV")
コンテキストについてはこちらを参考にしました。
ロギング
AWS Lambda Powertools Python
Pythonでログ出力する場合はLoggerを使いますが、ここではLambdaに特化したAWS Lambda Powertools PythonのLoggerを使います。
PowertoolsのLoggerを使うことでJSON形式の構造化されたログを残せます。
{
"level": "INFO",
"location": "collect.handler:7",
"message": "Collecting payment",
"timestamp": "2021-05-03 11:47:12,494+0200",
"service": "payment",
"cold_start": true,
"lambda_function_name": "test",
"lambda_function_memory_size": 128,
"lambda_function_arn": "arn:aws:lambda:eu-west-1:12345678910:function:test",
"lambda_request_id": "52fdfc07-2182-154f-163f-5f0f9a621d72",
"correlation_id": "correlation_id_value"
}
ログからはLambda関数のコールドスタートの状況やメモリサイズなど、パフォーマンスに影響する情報も確認できます。
Loggerには引数を渡すことができ、引数はoutputのmessageに出力されます。
logger.info({
"action": "do_something", # 引数を辞書型で渡すと構造化されて見やすい
"status": "run"
})
# output
# {
# "level": "INFO",
# "location": "collect.handler:7",
# "message": {
# "action": "do_something",
# "status": "run"
# },
# "timestamp": "2021-05-03 11:47:12,494+0200",
# "service": "payment",
# "cold_start": true,
# "lambda_function_name": "test",
# "lambda_function_memory_size": 128,
# "lambda_function_arn": "arn:aws:lambda:eu-west-1:12345678910:function:test",
# "lambda_request_id": "52fdfc07-2182-154f-163f-5f0f9a621d72",
# "correlation_id": "correlation_id_value"
# }
導入方法
Powertoolsの導入方法は2つあります。
- Lambda Layer として利用する
- PyPi からインストールする
ローカルでは開発用としてPyPiからインストールし、本番ではLambda Layerを使います。
ログの出力どころ
関数の開始と終了でそれぞれ出力します。
どの処理を開始したのか、どこまで終了したのかが追跡しやすくなります。
また、関数に渡す引数などのパラメーターがあればその値も併せて出力します。
def do_something(id: str):
logger.info({
"action": "do_something",
"id": id,
"status": "run"
})
print("do something")
logger.info({
"action": "do_something",
"id": id,
"status": "success"
})
テスト
pytestとmotoを使います。
pipenv install pytest --dev # テストフレームワーク / ローカルで使用するため本番環境には不要
pipenv install moto --dev # AWSモック / ローカルで使用するため本番環境には不要
motoを使うとBoto3を使った他のAWSリソースとの接続を再現できるため、確認のためにわざわざAWS上で動かす必要がなくなります。
pytestはフィクスチャーを使うことでテストの都度リソースの状態をリセットできます。
そのためテストの中でデータの状態が変わってしまうということは起きません。
フィクスチャーはconftest.pyに記載することで複数のテストコードと共有できます。
下記はDynamoDBにuserテーブルを都度用意するフィクスチャーの例です。
import boto3
from moto import mock_dynamodb2
import pytest
import json
from pathlib import Path
@pytest.fixture()
def aws_credentials():
"""Mocked AWS Credentials for moto.""" # 意図しないAWS環境のリソースを操作しないためのダミークレデンシャル
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
os.environ["AWS_DEFAULT_REGION"] = "ap-northeast-1"
@pytest.fixture()
def dynamodb(aws_credentials):
with mock_dynamodb2():
yield boto3.resource("dynamodb")
@pytest.fixture()
def user_table(dynamodb):
# Create Table
dynamodb.create_table(
TableName="user",
KeySchema=[{"AttributeName": "userId", "KeyType": "HASH"}],
AttributeDefinitions=[
{"AttributeName": "userId", "AttributeType": "S"},
],
BillingMode="PAY_PER_REQUEST",
)
# Put Test Data
user_table = dynamodb.Table("user")
input_data = Path(__file__).resolve().parent.joinpath("input_data", "user.json") # JSON形式のuserデータ ./input_data/user.json を指定している
with open(input_data) as f:
user_items = json.load(f)
with user_table.batch_writer() as batch:
for item in user_items:
batch.put_item(Item=item)
yield user_table
エラーハンドリング
if, elseやtry, exceptを使って処理を制御します。
どちらを使うべきかについてはどちらでも良いと思います。
ただしif, elseで複数の例外を制御するにはelse文を増やす必要があります。
else文の数によってはtry, exceptを使ったほうが簡潔に記述できるケースもあるため、内容に応じて選択するのが良いと思います。
例外は内容を分かりやすくするためにカスタム例外としてexceptions.pyに外だしします。
カスタム例外ではベースクラスとベースクラスを継承する例外クラスを定義します。
# ベースクラス
class FunctionError(Exception):
"""Base class for exceptions."""
pass
# ベースクラスを継承する例外クラス
class IDNotFound(FunctionError):
"""Exception raised if the ID is not found."""
def __init__(self, id):
self.user_id = user_id
def __str__(self):
return f"ID {self.user_id} is not found."
下記は判定処理の例です。
if user_id is None:
raise IDNotFound(user_id)
# output
# ID XXXXX is not found.
boto3のエラーはClientErrorで拾います。
下記のようにimportして使用します。
from botocore.exceptions import ClientError
def add_user(item: dict) -> str:
try:
ret = user_table.put_item(Item=item)
except ClientError as e:
raise logger.error(e)
else:
return ret["Item"]["userId"]
まとめ
自分がPythonでLambdaを作るときに意識する内容をざっと書き出しました。
こんなのあるんだなぁ、という発見があれば幸いです。