AWS DynamoDBからデータを取り出してJSON形式で出力するPythonのプログラム
Contents
DynamoDB
AWS(Amazon Web Services）のDynamoDBからデータを取り出して全件JSON形式て出力するPythonのプログラムを備忘録として記事にしておく。
尚、読み込むDynamoDBは以前に「ラズパイZEROとモーションセンサーで見守りシステムをつくる」の記事で作成したモーションセンサーでのカウント記録が格納されたテーブルとした。
大まかな手順は以下の通り。
- AWS IAMでDynamoDBにアクセスできるユーザを作成する
- Pythonでプログラムを作成する
関連記事
IAMでユーザを作成する
AWSのコンソールにログインをしてサービスからIAM（Identity and Access Management）を選択して左のメニューから”アクセス管理”ー＞”ユーザ”をクリックする。
ユーザを追加
上部の「ユーザを追加」ボタンをクリックする。
ユーザ詳細の設定で、
- ユーザ名：任意の文字列
- アクセスの種類：プログラムによるアクセスにチェックする
「次のステップ：アクセス権限」ボタンをクリックする。
アクセス権限
「既存のポリシーを直接アタッチ」を選択して、”AmazonDynamoDBReadOnlyAccess”※にチェックを入れて「次のステップ：確認」ボタンをクリックする。
※今回は参照するだけなのでこの権限でOK
管理ポリシーでAmazonDynamoDBReadOnlyAccessが追加されている事を確認して「アクセス権限の追加」ボタンをクリックする。
タグの追加
必要に応じてタグを追加する。
自分はタグは追加せずに「次のステップ：確認」ボタンをクリックした。
ユーザの作成
確認画面が表示されるので「ユーザの作成」ボタンを押すとユーザが作成されてアクセスキーIDとシークレットアクセスキーが表示されるので後でプログラム中で使用するので両方共に控えておく。
以上でユーザの作成は終了。
必要なモジュール
PythonアプリケーションからAWSの各種サービスを使うためのAWS SDK for Python (Boto3)をインストールする。
Boto3に関してはこちらのページを参照。
自分はWindows環境にAnacondaでPython環境を構築しているのでAnacondaプロンプトから以下のコマンドでBoto3をインストールした。
pip install boto3
プログラム
プログラム実行環境
OS | Windows10 Pro及びHome 64Bit |
Python | Ver3.7.7（Anaconda上で実行） |
プログラムソースコード
プログラムは以下の通り。
Developers.ioのこちらの記事を参考にしている。
というか中身はそのままで自分の理解の為にコメントを書き加えただけ。
# -*- 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 boto3
from boto3.resources.base import ServiceResource
from boto3.dynamodb.types import Binary
def scan_table(table_name: str, dynamodb: ServiceResource) -> List[dict]:
table = dynamodb.Table(table_name) # DynamoDBテーブル読み込み
resp = table.scan() # scan
table_items = resp['Items']
while 'LastEvaluatedKey' in resp: # １回のscanでは取得できる件数に限界があるので繰り返す
resp = table.scan(
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 Name' # DynamoDBのエーブル名
dynamodb = boto3.resource('dynamodb',
aws_access_key_id='XXXXXX', # 控えておいたアクセスキーID
aws_secret_access_key='XXXXXX', # 控えておいたシークレットアクセスキー
region_name='ap-northeast-1') # リージョン（東京）
table_items = scan_table(table_name, dynamodb)
put_data(table_items)補足説明
１６～２５行目
DynamoDBをtable.scanで読み込んでいる。
１度scanした後、whileでscanを繰り返しているのは１度のscanで読み取れる件数に限界があるのでデータの最後まで繰り返している。
LastEvaluatedKeyに取得できなかったデータの最初の位置がセットされる。
２７～４２行目
DynamoDBから取り出した値をそのまま出力しようとするとTypeErrorが発生してしまうので型変換を行っている。
例えば数値（Number型）の項目はDecimal型でと取り出されるのでInt型かfloat型に変換している。
４６行目のjson.dumpでの出力時のdefault指定でこの関数を呼び出して型変換を行っている。
４６行目
JSON形式で出力する。
ensure_ascii=False：非 ASCII 文字はエスケープせずにそのまま出力する
indent=4：４Byteの空白でインデントする
終わりに
以上でDynamoDBから全件JSON形式でデータを取り出すプログラムの記事は終了。
いずれキー指定で取り出すプログラムも作ってみたい。
この記事が何処かで誰かの役に立つことを願っている。
尚、当記事中の商品へのリンクはAmazonアソシエイトへのリンクが含まれています。Amazonのアソシエイトとして、当メディアは適格販売により収入を得ていますのでご了承ください。
最近のコメント