AWSサービス連携① 〜Lambda/python〜

業務でAWS/Lambdaを用いてのサーバレス構成を構築する機会があったので、実装方法などを備忘録がてら残しておくことにした。開発言語はpython。

boto3とは
DynamoDB
まとめ

boto3とは

pythonでAWSのサービスを利用したいときに使う主流のライブラリ。この記事では以下の使い方についてまとめた。

DynamoDB
RDS
S3
SNS
Lambda

DynamoDB

レコードを書き込む

dynamo = boto3.resource('dynamodb', region_name='XXXXXX')
table = dynamo.Table('table_name')
response = table.put_item(Item=data)

Dynamoを扱う時に注意すべきことがある。それはDynamoで保持する数字型データはDecimal型という点だ。そのため登録するデータのうちfloat型がある場合はDecimal型に変換する必要があるし、（intの場合は自動的に変換される）取得するデータもDecimalであるということを気に留めておかなくてはならない。この時、さらに注意しなければならない点が2つある。（筆者はそこそこハマってしまった。。。）1つがnumpy.float64の存在である。pandasなんかを使って計算していると不意に現れてくる憎きこの型。floatとは別の型として扱われるので注意が必要である。2つ目は少数点を含む場合。

if type(data) is float:
   data = Decimal(data)

上記のような変換を行っていたところ、data=0.1の時に

 "errorMessage": "[<class 'decimal.Inexact'>, <class 'decimal.Rounded'>]"

とよくわからないエラーを吐いた。解決方法を調べて以下のようにして解決した。

ctx = Context(prec=38)
value = ctx.create_decimal_from_float(value)

floatの有効桁数関連で起きていた模様。詳しくはググってください🙏precの値は任意でよいはず。同じようなエラーが出た時は試してほしい。

レコードを取り出す

dynamo = boto3.resource('dynamodb', region_name='XXXXXX')
table = dynamo.Table('table_name')
response = table.query(XXXXXX)  # クエリを指定する

取り出す時はqueryメソッドを使用して該当のレコードを取得する。その時に指定できるものをいくつか挙げる。

KeyConditionExpression：　パーティション・ソートキーにのみ使用でき、Keyで指定する。
FilterExpression：　上記以外のキーを使いたい場合に使用し、Attrで指定する。
ScanIndexForward：　ScanIndexForward=Falseで最新データから順に取得する。
Limit：　データをいくつ取得するかを示す。ScanIndexForwardと組み合わせることが多い。

id = 10（パーティションキー）, food = “寿司”（通常キー）, 最新データ5件取得の場合を下記例を示す。

response = table.query(
　　　　　　　　　　　　　　　KeyConditionExpression=Key('id').eq(10) ,
                FilterExpression=Attr('food').eq("寿司"),
                ScanIndexForward=False,
                Limit=5,
)

クエリはeqやne(not equal), gte(grater than and equal), between等の条件式を使用する。指定できる条件式についてはhttps://boto3.amazonaws.com/v1/documentation/api/latest/reference/customizations/dynamodb.htmlを参考にしてほしい。またAND条件やOR条件も設定可能で、&や|を用いる。この場合は以下のように記載する。

options = {　
         KeyConditionExpression=Key('id').eq(10) ,
         FilterExpression=Attr('food').eq("寿司") & Attr('age').gte(20)
}
response = table.query(**options)

レコードを取り出す時にも注意点がある。それは、一度のクエリで取得できるデータは1MBまでであるという点だ。もし1MB以上のデータを取得したい場合は、ページング処理を実装する必要がある。

クエリに対して全件取得できなかった時には「LastEvaluatedKey」というキーが含まれていて、次点の開始位置を示すキーが格納されている。これを利用して、このキーがなくなるまで繰り返しクエリを投げれば良い。クエリに「ExclusiveStartKey」という開始位置を示すオプションを指定できるので、下記のようにすればよい。

while "LastEvaluatedKey" in result:
          start_key = result["LastEvaluatedKey"]
          options = {　
                        KeyConditionExpression=Key('id').eq(10) ,
                        FilterExpression=Attr('food').eq("寿司") & Attr('age').gte(20),
                        ScanIndexForward=False,
                        ExclusiveStartKey=start_key
           }
           result = table(**options)