DynamoDB上のデータをファイルにエクスポートする際のJSONシリアライズについて考えてみる（スクリプト付き）

import json
import boto3

dynamodb = boto3.resource('dynamodb')
table_name = 'sample-table'
table = dynamodb.Table(table_name)

resp = table.scan()
items = resp['Items']

for item in items:
    for kv in item.items():
        _, v = kv
        print(kv, type(v))

$ python scan.py
('k01', Decimal('1234567890')) <class 'decimal.Decimal'>
('k02', 'テスト') <class 'str'>
('k03', Binary(b'Binary Data')) <class 'boto3.dynamodb.types.Binary'>
('k04', True) <class 'bool'>
('k05', None) <class 'NoneType'>
('k06', ['val1', Decimal('9876543210')]) <class 'list'>
('k07', {'key': 'hoge'}) <class 'dict'>
('k08', {'bar', 'foo'}) <class 'set'>
('k09', {Decimal('222'), Decimal('111')}) <class 'set'>
('k10', {Binary(b'Binary Set 2'), Binary(b'Binary Set 1')}) <class 'set'>

import json
from decimal import Decimal
from boto3.dynamodb.types import Binary

def default(obj) -> object:
    if isinstance(obj, Decimal):
        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()
    elif isinstance(obj, set):
        return list(obj)
    try:
        return str(obj)
    except Exception:
        return None

json.dump(<対象データ>, <書き込み先のファイル>, default=default)

import json
import sys
from typing import List
from decimal import Decimal

import boto3
from boto3.resources.base import ServiceResource
from boto3.dynamodb.types import Binary

def main(table_name: str, dynamodb_resource: ServiceResource) -> None:
    table_items = scan_table(table_name, dynamodb_resource)
    put_data(table_items)

def scan_table(table_name: str, dynamodb_resource: ServiceResource) -> List```dict:
    table = dynamodb_resource.Table(table_name)
    resp = table.scan()
    table_items = resp['Items']
    while 'LastEvaluatedKey' in resp:
        resp = table.scan(
            ExclusiveStartKey=resp['LastEvaluatedKey']
        )
        table_items.extend(resp['Items'])
    return table_items

def default(obj) -> object:
    if isinstance(obj, Decimal):
        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, ensure_ascii=False, sort_keys=True, indent=4)

if __name__ == '__main__':
    table_name = 'sample-table'
    dynamodb_resource = boto3.resource('dynamodb')

    main(table_name, dynamodb_resource)

[
    {
        "k01": 1234567890,
        "k02": "テスト",
        "k03": "Binary Data",
        "k04": true,
        "k05": null,
        "k06": [
            "val1",
            9876543210
        ],
        "k07": {
            "key": "hoge"
        },
        "k08": [
            "foo",
            "bar"
        ],
        "k09": [
            222,
            111
        ],
        "k10": [
            "Binary Set 1",
            "Binary Set 2"
        ]
    }
]

def default(obj) -> object:
    return obj

$ python script.py
Traceback (most recent call last):
  File "script.py", line 41, in <module>
    main(table_name, dynamodb_resource)
  File "script.py", line 13, in main
    put_data(table_items)
  File "fetcher.py", line 34, in put_data
    json.dump(table_items, f, default=default, ensure_ascii=False, sort_keys=True, indent=4)
  File "/usr/lib64/python3.6/json/__init__.py", line 179, in dump
    for chunk in iterable:
  File "/usr/lib64/python3.6/json/encoder.py", line 428, in _iterencode
    yield from _iterencode_list(o, _current_indent_level)
  File "/usr/lib64/python3.6/json/encoder.py", line 325, in _iterencode_list
    yield from chunks
  File "/usr/lib64/python3.6/json/encoder.py", line 404, in _iterencode_dict
    yield from chunks
  File "/usr/lib64/python3.6/json/encoder.py", line 438, in _iterencode
    yield from _iterencode(o, _current_indent_level)
  File "/usr/lib64/python3.6/json/encoder.py", line 435, in _iterencode
    raise ValueError("Circular reference detected")
ValueError: Circular reference detected