PythonでのAWS Lambda共通DynamoDB操作Util

python での DBUtil の作成方法

以下に、Python で AWS Lambda 内で共通的に使用される DynamoDB 操作のためのユーティリティクラスの例を示します。このクラスを使用すると、DynamoDB テーブルへのアクセスや操作が容易になります。

import boto3

dynamodb = boto3.resource('dynamodb', endpoint_url="XXXXXXX")


class DynamoDB:
    """DynamoDBの操作(CRUD)クラス
    Args:
        table_name: テーブル名
    Attributes:
        table: DynamoDBテーブルのオブジェクト
    """

    def __init__(self, table_name):
        try:
            self.table = dynamodb.Table(table_name)
        except Exception:
            raise

    def put_item(self, item_dict):
        """アイテム追加
        Args:
            item_dict: 追加情報
        Returns:
            追加結果(dict)
        """
        response = self.table.put_item(
            Item=item_dict
        )
        return response

    def update_item(self, key_dict, update_expression, expression_attribute_values, expression_attribute_names):
        """アイテム更新
        Args:
            key_dict: パーティションキー値（更新列指定）
            update_expression: 更新カラム指定
            expression_attribute_values: 更新属性値
            expression_attribute_names: 更新属性名
        Returns:
            更新結果(dict)
        """
        response = self.table.update_item(
            Key=key_dict,
            UpdateExpression=update_expression,
            ExpressionAttributeValues=expression_attribute_values,
            ExpressionAttributeNames=expression_attribute_names,
        )
        return response

    def delete_item(self, partition_key_value, sort_key_value=None):
        """アイテム削除
        Args:
            partition_key_value: パーティションキー値
            sort_key_value: ソートキー値
        Returns:
            削除結果(dict)
        """
        if sort_key_value is not None:
            response = self.table.delete_item(
                key=partition_key_value,
                sortkey=sort_key_value
            )
        else:
            response = self.table.delete_item(
                Key=partition_key_value
            )

        return response

    def get_item(self, item_dict):
        """アイテム一意検索
        Args:
            item_dict: 検索情報
        Returns:
            検索結果(dict)
        """
        response = self.table.get_item(
            Key=item_dict,
        )

        return response.get('Item')

    def exist_item(self, item_dict):
        """アイテム一意検索(存在チェックあり)
        Args:
            item_dict: 検索情報
        Returns:
            検索結果(dict)
            ヒットいない場合None
        """
        response = self.table.get_item(
            Key=item_dict,
        )

        if 'Item' in response:
            return True
        else:
            return False

    def query(self, key_condition_expression, filter_expression=None):
        """全データクエリ(LIMITなし)
        Args:
            key_condition_expression: キー検索条件
            filter_expression: 検索条件
        Returns:
            検索結果(dict)

        使い方)
        from boto3.dynamodb.conditions import Key, Attr
        key_condition_expression=Key('Id').eq('1')
        filter_expression=Attr('XXXXX').contains('○○')
        Item = DynamoDB('Table','Id').query(
            key_condition_expression, filter_expression)
        print(Item)
        """
        params = {'KeyConditionExpression': key_condition_expression}
        response = []
        if filter_expression is not None:
            params['FilterExpression'] = filter_expression
        while True:
            result = self.table.query(**params)
            response.extend(result['Items'])
            if 'LastEvaluatedKey' not in result:
                break
            params['ExclusiveStartKey'] = result['LastEvaluatedKey']
        return response


def scan(self, filter_expression):
    """全データスキャン(LIMITなし)
        Args:
            filter_expression: 検索条件
        Returns:
            検索結果(dict)

        使い方)
        from boto3.dynamodb.conditions import Key, Attr
        filter_expression=Attr('XXXXX').eq('○○')
        Items = DynamoDB('Table','Id').scan(filter_expression)
        """
    params = {}
    response = []
    if filter_expression is not None:
        params['FilterExpression'] = filter_expression
    while True:
        result = self.table.scan(**params)
        response.extend(result['Items'])
        if 'LastEvaluatedKey' not in result:
            break
        params['ExclusiveStartKey'] = result['LastEvaluatedKey']
    return response

上記の例では、DynamoDB というクラスが定義されています。このクラスは DynamoDB テーブルへの共通的な操作を提供します。

__init__メソッドでは、DynamoDB リソースとテーブルを初期化します。table_name パラメータは、操作する対象の DynamoDB テーブルの名前です。

以下のメソッドが提供されています:

get_item: 指定されたキーを使用してアイテムを取得します。
put_item: アイテムをテーブルに追加します。
update_item: アイテムを更新します。アップデート式と式の属性値を指定する必要があります。
delete_item: 指定されたキーを持つアイテムを削除します。
exist_item: 指定されたキーを持つアイテムが存在するかを判断します。
query: 指定されたキーと検索条件によりデータを取得します。
scan: 指定された検索条件によりデータを取得します。

これらのメソッドは、必要に応じてカスタマイズや追加の操作を行うことができます。また、AWS SDK for Python (Boto3)の機能を活用して、他の DynamoDB 操作も実装できます。
このクラスを使用するには、Lambda 関数内でインスタンスを作成し、適切なメソッドを呼び出すことができます。
例えば、以下のような使い方ができます:

def lambda_handler(event, context):
    dynamodb_utils = DynamoDBUtils('my-table-name')

    # アイテムの取得
    key = {'id': '123'}
    item = dynamodb_utils.get_item(key)
    print(item)

    # アイテムの追加
    new_item = {'id': '456', 'name': 'John'}
    response = dynamodb_utils.put_item(new_item)
    print(response)

    # アイテムの更新
    update_expression = 'SET #attrName = :attrValue'
    expression_attribute_values = {':attrValue': 'updated value'}
    response = dynamodb_utils.update_item(key, update_expression, expression_attribute_values, None)
    print(response)