DynamoDBのテーブルをscanするPythonコード

概要

• boto3(AWS SDK for Python)でDynamoDBをスキャンするコードです
• きちんとscanするために LastEvaluatedKey を使ってループする必要があります
• 自分用スニペットです

コード

def get_records(table, **kwargs):
    while True:
        response = table.scan(**kwargs)
        for item in response['Items']:
            yield item
        if 'LastEvaluatedKey' not in response:
            break
        kwargs.update(ExclusiveStartKey=response['LastEvaluatedKey'])

使い方サンプル

フルスキャン

import boto3

dynamodb = boto3.resource('dynamodb')
records = get_records(
    dynamodb.Table('mytable')
)
for record in records:
    print(record)

フルスキャン,Projection指定

import boto3

dynamodb = boto3.resource('dynamodb')
records = get_records(
    dynamodb.Table('mytable'),
    ProjectionExpression='MyPartitionKey, MyTimestamp'
)
for record in records:
    print(record)

Projection指定,Filter指定

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
records = get_records(
    dynamodb.Table('mytable'),
    ProjectionExpression='MyPartitionKey, MyTimestamp',
    FilterExpression=Key('MyTimestamp').begins_with('2020-02-23')
)
for record in records:
    print(record)

Projection指定,Filter指定,Index指定

import boto3
from boto3.dynamodb.conditions import Key

dynamodb = boto3.resource('dynamodb')
records = get_records(
    dynamodb.Table('mytable'),
    ProjectionExpression='MyPartitionKey, MyTimestamp',
    FilterExpression=Key('MyTimestamp').begins_with('2020-02-23'),
    IndexName='MyTimestamp-index'
)
for record in records:
    print(record)

リンク

• https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/dynamodb.html#DynamoDB.Table.scan
• https://boto3.amazonaws.com/v1/documentation/api/latest/_modules/boto3/dynamodb/conditions.html
• https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/GettingStarted.Python.04.html