LambdaでキーのみのDynamoDBテーブルを作成させてみた

業務で色々とDynamoDBをもてあそぶ機会があるんですが、DynamoDBのテーブル設定するのが面倒くさいのとキーのところのコピペがうまくいかなかったりと、なんだかイライラとするところが多くて、これ何とかなんないのかと思って、少し調べました。
既にデータをインポートするところは作っているのでキーだけを付与するDynamoDB作成用のツールがほしかったわけです。ググってコピペって完成です。いつも通り

とりあえず個人で使用するので環境変数を書き替えてLambdaのコンソールで[テスト]ってすれば動くようになっています。後は単独ハッシュキーのプライマリキーかハッシュ+ソートの複合プライマリーキーを設定して作成できるようにしています。セカンダリ―インデックスは使用しないので今回は対象外です。そしてオンデマンドです。
エラー処理はほぼないです。相変わらずのへぼちんです。そして追加ライブラリはなしです。一応、Lambdaのpython3.8ランタイムで動きました。

■ソースコード

import datetime
import logging
import json
import os
import urllib

import boto3

LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)
dynamodb = boto3.resource('dynamodb')

# DynamoDB
def make_table(table_name,key_name,tag_value):
    try:
        if type(key_name) is str:         
           attrdef  = [{ 'AttributeName': key_name ,'AttributeType': 'S'}]
           keyschem = [{ 'AttributeName': key_name ,'KeyType': 'HASH'}]
        elif len(key_name) == 2:
           attrdef  = [
                        { 'AttributeName': key_name[0] ,'AttributeType': 'S'},
                        { 'AttributeName': key_name[1] ,'AttributeType': 'S'}
                      ]
           keyschem = [
                        { 'AttributeName': key_name[0] ,'KeyType': 'HASH'},
                        { 'AttributeName': key_name[1] ,'KeyType': 'RANGE'}
                      ]
        else:
           return False

        tagskeyval = {'Key': 'Billing','Value': tag_value}
        response = boto3.client('dynamodb').create_table(
            AttributeDefinitions=attrdef,
            TableName=table_name,
            KeySchema=keyschem,
            BillingMode='PAY_PER_REQUEST',
            Tags=[tagskeyval]
        )
        LOGGER.info("Completed registration")
        return True
    except Exception as e:
        LOGGER.error(e)
        raise e

# SNS
def sendmessage(subject,msg):
    try:
        sns_arn = os.getenv('SNS_ARN')
        client = boto3.client('sns')
        request = {
                     'TopicArn': sns_arn,
                     'Message': msg,
                     'Subject': subject
                  }
        r = client.publish(**request)
        return r
    except Exception as e:
        LOGGER.error(e)
        raise e 

# main
def lambda_handler(event, context):
    try:
        st = datetime.datetime.now()
        mg = u'◆作成テーブル：\n'
        tablename = os.getenv('TABLE_NAME')
        tagvalue = os.getenv('TAG_VALUE')
        #keyname = 'base_search_key'
        k1 = os.getenv('KEY_NAME1')
        k2 = os.getenv('KEY_NAME2')
        str_keyname =k1
        if len(k2) != 0:
           list_keyname =[k1,k2]
           stat = make_table(tablename,list_keyname,tagvalue)
        else:
           stat = make_table(tablename,str_keyname,tagvalue)

        LOGGER.info("Completed registration")
        mg= mg + tablename + '\n'
        sb=u'処理終了'
        end = datetime.datetime.now()
        sendmessage(sb,mg)
        return True
    except Exception as e:
        LOGGER.error(e)
        raise e

■参考サイト
https://michimani.net/post/aws-operate-dynamodb-by-python/#%E3%83%86%E3%83%BC%E3%83%96%E3%83%AB%E3%81%AE%E4%BD%9C%E6%88%90
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Client.create_table
https://www.yamamanx.com/dynamodb-daily-create-delete/

これでちまちまテーブルを作らなくてよくなったので少しだけ楽になったかなぁと思いました。