この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。
boto3 で楽しむ AWS PythonLife 一人AdventCalendarです。
boto3のドキュメントを通して、サービス別にどういった事が出来るのかを理解したり、管理コンソールを通さずにTerminalだけで完結できるように検証していくことが目的になります。
14日目はLambdaの関数作成及び更新をやってみます。
目次
boto3を通してLambdaでできること
ドキュメントは下記リンク先です。
ざっくりと以下のことができます。
- 関数の操作(作成・削除・更新)
- 関数のエイリアス操作(作成・削除・更新)
作業リージョン
- Lambda関数を作成する際に利用するS3のリージョンと合わせる必要がある
- boto3経由でのS3リージョン指定はus-east-1に明示する必要がある
上記の2点の都合で、Lambda関数の作成リージョンはus-east-1になります。
関数用パッケージの用意
Lambda関数作成に用いるパッケージを用意する必要がありますが、Lambdaを新規作成した際に追加されるサンプルスクリプトでも構いません。
サンプルスクリプトを追加することが手間な場合は、英数字名でフォルダを作成し、その中に以下の内容で準備してください。
lambda_function.py
import json
def lambda_handler(event, context):
# TODO implement
return {
'statusCode': 200,
'body': json.dumps('Hello from Lambda!')
}今回の手順
用意しておいた関数用パッケージを圧縮してS3に転送して、Lambda上に転送したファイルをもとに関数を作成します。
- 関数用パッケージをZip圧縮
- 圧縮したファイルをS3へ転送
- Lambda関数をS3へ転送したZipファイルをもとに作成
実行手順
% python main.py
Input Profile name [default]>>
Select action
[0] create
[1] update
>> 0
Select bucket
[0] XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX
>> 0
Input package_name >> 14
Select role arn
[0] arn:aws:iam::XXXXXXXXXXXXXXX:role/XXXXXXXXXXXX
>> 0
Input function name >> test
Input handler >> lambda_function.lambda_handlermain.py
import boto3
from zipfile import ZipFile
from pprint import pprint
class LambdaWizard:
_client_name = 'lambda'
_session = None
_bucket_name = None
_role_arn = None
_package_name = None
_function_name = None
_handler = None
def __init__(self, profile_name):
self._session = boto3.Session(profile_name=profile_name, region_name='us-east-1')
@property
def session(self):
return self._session
def get_client(self, client_name=None, region_name=None):
if not client_name:
client_name = self._client_name
params = {}
if region_name:
params['region_name'] = region_name
return self.session.client(client_name, **params)
@property
def client_name(self):
return self._client_name
@property
def bucket_name(self):
return self._bucket_name
@property
def role_arn(self):
return self._role_arn
@property
def package_name(self):
return self._package_name
@property
def package_zip_name(self):
return "{}.zip".format(self._package_name)
@property
def function_name(self):
return self._function_name
@property
def handler(self):
return self._handler
def get_role_list(self):
roles = self.get_client('iam').list_roles()
role_arns = list()
for role in roles['Roles']:
role_arns.append(role['Arn'])
return role_arns
def get_bucket_list(self):
bucket_list = self.get_client('s3').list_buckets()
bucket_names = list()
for bucket in bucket_list['Buckets']:
bucket_names.append(bucket['Name'])
return bucket_names
def create_function(self):
params = {
'FunctionName': self.function_name,
'Runtime': 'python3.7',
'Role': self.role_arn,
'Handler': self.handler,
'Code': {
'S3Bucket': self.bucket_name,
'S3Key': self.package_zip_name,
}
}
return self.get_client().create_function(**params)
def update_function(self):
params = {
'FunctionName': self.function_name,
'S3Bucket': self.bucket_name,
'S3Key': self.package_zip_name,
}
return self.get_client().update_function_code(**params)
def upload_zip(self):
with ZipFile(self.package_zip_name, 'w') as zip_handler:
zip_handler.write("{}/lambda_function.py".format(self.package_name))
params = {
'Filename': "./{}".format(self.package_zip_name),
'Bucket': self.bucket_name,
'Key': self.package_zip_name
}
return self.get_client('s3').upload_file(**params)
def prompt_list_base(self, target_list, message):
name = None
while True:
print('\n{}'.format(message))
for name in target_list:
print('[{}] {}'.format(target_list.index(name), name))
index = input('>> ')
if len(index) != 0 and int(index) < len(target_list):
name = target_list[int(index)]
break
return name
def prompt_bucket(self):
self._bucket_name = self.prompt_list_base(self.get_bucket_list(), 'Select bucket')
def prompt_role_arn(self):
self._role_arn = self.prompt_list_base(self.get_role_list(), 'Select role arn')
def prompt_action(self):
return self.prompt_list_base(['create', 'update'], 'Select action')
def prompt_base(self, message):
value = None
while True:
value = input("\n{} >> ".format(message))
if value and len(value) != 0:
break
return value
def prompt_package_name(self):
self._package_name = self.prompt_base('Input package_name')
def prompt_function_name(self):
self._function_name= self.prompt_base('Input function name')
def prompt_handler(self):
self._handler = self.prompt_base('Input handler')
@staticmethod
def prompt():
default_profile_name = 'default'
profile_name = input('Input Profile name [{}]>> '.format(default_profile_name))
if len(profile_name) == 0:
profile_name = default_profile_name
wizard = LambdaWizard(profile_name)
action = wizard.prompt_action()
wizard.prompt_bucket()
wizard.prompt_package_name()
wizard.prompt_role_arn()
wizard.prompt_function_name()
wizard.upload_zip()
if action == 'create':
wizard.prompt_handler()
pprint(wizard.create_function())
else:
pprint(wizard.update_function())
if __name__ == '__main__':
LambdaWizard.prompt()まとめ
ZipFileを直接上げる検証を進めていましたが、文字コードの問題が解決へ至れなかった関係でS3からの検証に変更しました。
ZipFile圧縮するリソースをgitリポジトリに且つ対象ファイルを適切にフィルタリングすることで、適切なバージョン管理も可能になると思われます。
16
