DynamoDBからfilter指定でデータを取り出す Pythonのプログラム

DynamoDB

以前にDynamoDBから全件データをJSON形式で取り出すプログラムに関する記事を書いた。

今回は全件では無く、指定した範囲・値のデータをDynamoDBから取り出すプログラムについて分かった事を記事にしておく。

事前準備

アクセス用のアクセスキーIDの作成やPythonからDynamoDBにアクセスするために必要なモジュール（Boto3)のインストールについては前回の記事を参照。

プログラム

プログラム実行環境

プログラムソースコード

scan時にフィルターを指定するプログラムは以下の通り。

# -*- coding: utf-8 -*-
"""
Created on Mon Jun 29 14:55:27 2020

@author: Souichirou.Kikuchi
"""

import json
from typing import List
from decimal import Decimal
import datetime as dt

import boto3
from boto3.resources.base import ServiceResource
from boto3.dynamodb.types import Binary
from boto3.dynamodb.conditions import Key, Attr # Query用

def scan_table(table_name: str, dynamodb: ServiceResource) -> List[dict]:
    table = dynamodb.Table(table_name) # DynamoDBテーブル読み込み
    resp = table.scan(
        # FilterExpression=Attr('CountTimeFrom').eq('2020-03-29 10:30:00')
        # FilterExpression=Attr('CountTimeFrom').gt('2020-07-04 10:30:00')
        # FilterExpression=Attr('CountTimeFrom').contains('2020-03-29 10:40:00')
        # FilterExpression=Key('CountTimeFrom').eq('2020-03-29 10:30:00')
        FilterExpression=Key('CountTimeFrom').between('2020-03-29 10:30:00','2020-03-29 10:40:00')
        )
    table_items = resp['Items'] # 1回のTableScanで取得できる件数に限界があるので複数回に渡って取得する
    while 'LastEvaluatedKey' in resp: # 最後まで読み込み
        resp = table.scan(
            FilterExpression=Key('CountTimeFrom').between('2020-03-29 10:30:00','2020-03-29 10:40:00'),
            ExclusiveStartKey=resp['LastEvaluatedKey']
        )
        table_items.extend(resp['Items'])
    return table_items

def default_proc(obj) -> object: # 型変換
    if isinstance(obj, Decimal): # Decimalであればintかfloatに変換
        if int(obj) == obj:
            return int(obj)
        else:
            return float(obj)
    elif isinstance(obj, Binary):
        return obj.value
    elif isinstance(obj, bytes):
        return obj.decode(encoding='default')
    elif isinstance(obj, set):
        return list(obj)
    try:
        return str(obj)
    except Exception:
        return None

def put_data(table_items: List[dict]) -> None:
    with open('./export.json', 'w') as f:
        json.dump(table_items, f, default=default_proc, ensure_ascii=False, sort_keys=True, indent=4)

if __name__ == '__main__':
    table_name = 'Table名'
    dynamodb = boto3.resource('dynamodb',
                          aws_access_key_id='アクセスキー',
                          aws_secret_access_key='シークレットアクセスキー',
                          region_name='ap-northeast-1') # リージョン
    table_items = scan_table(table_name, dynamodb)
    put_data(table_items)

補足説明

２５行目はプライマリキーのCountTimeFromに対して範囲指定でフィルターを掛けた例。

その他には２１～２４行目の様な指定方法も可能。

• eq:イコール
• gt:大なり
• gte：大なりイコール
• contains：含む
• lt：小なり
• lte：小なりイコール

以上で今回の記事は終了とする。

この記事が何処かで誰かの役に立つことを願っている。

尚、当記事中の商品へのリンクはAmazonアソシエイトへのリンクが含まれています。Amazonのアソシエイトとして、当メディアは適格販売により収入を得ていますのでご了承ください。