[Python, Boto3] DynamoDBのクエリで1MBを超えるデータをすべて取得する関数のスニペット

import boto3
from boto3.dynamodb.conditions import Key, Attr

def query_all_items(table_name, key_condition_expression, filter_expression=None):
    """
    DynamoDBテーブルからすべてのアイテムをクエリし、1MBを超えるデータも取得する。

    :param table_name: クエリするテーブルの名前
    :param key_condition_expression: クエリのKeyConditionExpression
    :param filter_expression: オプションのFilterExpression
    :return: すべてのアイテムのリスト
    """
    # DynamoDBクライアントの初期化
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)

    # すべてのアイテムを格納するリスト
    all_items = []
    last_evaluated_key = None

    # ページングしてクエリを実行
    while True:
        query_params = {
            "KeyConditionExpression": key_condition_expression
        }
        # FilterExpressionがある場合は追加
        if filter_expression:
            query_params["FilterExpression"] = filter_expression

        # LastEvaluatedKeyがある場合はExclusiveStartKeyを設定
        if last_evaluated_key:
            query_params["ExclusiveStartKey"] = last_evaluated_key

        # クエリの実行
        response = table.query(query_params)

        # 結果をリストに追加
        all_items.extend(response['Items'])

        # LastEvaluatedKeyが存在する場合、次のページがある
        last_evaluated_key = response.get('LastEvaluatedKey')

        # 次のページがない場合は終了
        if not last_evaluated_key:
            break

    return all_items

import boto3
from boto3.dynamodb.conditions import Key, Attr

def query_all_items(table_name, key_condition_expression, filter_expression=None):
    """
    DynamoDBテーブルからすべてのアイテムをクエリし、1MBを超えるデータも取得する。

    :param table_name: クエリするテーブルの名前
    :param key_condition_expression: クエリのKeyConditionExpression
    :param filter_expression: オプションのFilterExpression
    :return: すべてのアイテムのリスト
    """
    # DynamoDBクライアントの初期化
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table(table_name)

    # すべてのアイテムを格納するリスト
    all_items = []
    last_evaluated_key = None

    # ページングしてクエリを実行
    while True:
        query_params = {
            "KeyConditionExpression": key_condition_expression
        }
        # FilterExpressionがある場合は追加
        if filter_expression:
            query_params["FilterExpression"] = filter_expression

        # LastEvaluatedKeyがある場合はExclusiveStartKeyを設定
        if last_evaluated_key:
            query_params["ExclusiveStartKey"] = last_evaluated_key

        # クエリの実行
        response = table.query(query_params)

        # 結果をリストに追加
        all_items.extend(response['Items'])

        # LastEvaluatedKeyが存在する場合、次のページがある
        last_evaluated_key = response.get('LastEvaluatedKey')

        # 次のページがない場合は終了
        if not last_evaluated_key:
            break

    return all_items

# クエリ条件の定義（例: 'partition_key' が 'example_value' と等しい）
key_condition = Key("partition_key").eq("example_value")

# すべてのアイテムを取得
items = query_all_items("your_table_name", key_condition)

# 結果の出力
print(items)

# クエリ条件の定義（例: 'partition_key' が 'example_value' と等しい）
key_condition = Key("partition_key").eq("example_value")

# すべてのアイテムを取得
items = query_all_items("your_table_name", key_condition)

# 結果の出力
print(items)

# クエリ条件の定義
key_condition = Key("partition_key").eq("example_value")

# フィルタ条件の定義（例: 属性 'status' が 'active' のものだけ取得）
filter_condition = Attr('status').eq('active')

# すべてのアイテムを取得
items = query_all_items("your_table_name", key_condition, filter_condition)

# 結果の出力
print(items)

# クエリ条件の定義
key_condition = Key("partition_key").eq("example_value")

# フィルタ条件の定義（例: 属性 'status' が 'active' のものだけ取得）
filter_condition = Attr('status').eq('active')

# すべてのアイテムを取得
items = query_all_items("your_table_name", key_condition, filter_condition)

# 結果の出力
print(items)