[Boto3 Adv-Cal DAY14]Lambdaの関数作成と更新をやってみた

boto3 で楽しむ AWS PythonLife 一人AdventCalendarです。

boto3のドキュメントを通して、サービス別にどういった事が出来るのかを理解したり、管理コンソールを通さずにTerminalだけで完結できるように検証していくことが目的になります。

14日目はLambdaの関数作成及び更新をやってみます。

目次

• boto3を通してLambdaでできること
• 作業リージョン
• 関数用パッケージの用意 - lambda_function.py
• 今回の手順
  • 実行手順
  • main.py
• まとめ

boto3を通してLambdaでできること

ドキュメントは下記リンク先です。

• Boto3 document - 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上に転送したファイルをもとに関数を作成します。

1. 関数用パッケージをZip圧縮
2. 圧縮したファイルをS3へ転送
3. 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_handler

main.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リポジトリに且つ対象ファイルを適切にフィルタリングすることで、適切なバージョン管理も可能になると思われます。