こんにちは。キーボードの悪魔です。
LOFREE FLOWという、ロープロファイルでは初めてガスケットマウントを採用したキーボードを購入してみました。
なかなかに素敵なうち心地で驚いています。食べ物で表現するならしっとりしたティラミスでしょうか。
今回のテーマはBoto3で取得したレスポンスのフィールドにアクセスする方法についてです。
あるS3バケットの特定プリフィックスのオブジェクトのキーを取得する関数を題材にコードを書いてみましょう。
良くない書き方
getObjectKeys関数を書いてみました。ページイテレータの’Contents’フィールドにアクセスすることで、各コンテンツの’Key’を取得し、リストとして返します。
import argparse
from boto3.session import Session
from mypy_boto3_s3 import S3Client
session = Session()
def getObjectKeys(bucket: str, prefix: str) -> list[str]:
client: S3Client = session.client('s3')
paginator = client.get_paginator('list_objects_v2')
return [
content['Key']
for page in paginator.paginate(Bucket=bucket, Prefix=prefix)
for content in page['Contents']
]
def main():
parser = argparse.ArgumentParser()
parser.add_argument('bucket', help='name of the s3 bucket.')
parser.add_argument('prefix', help='prefix for listing s3 objects.')
args = parser.parse_args()
bucket = args.bucket
prefix = args.prefix
for key in getObjectKeys(bucket, prefix):
print(key)
if __name__ == '__main__':
main()
実行してみましょう。まずはプリフィックスについて指定せずに、全オブジェクトのキーを取得してみます。
エラーもなくキーが取得できています。期待した動作です。
$ python s3_getObjectKeys_bad.py koha-test '' AWSLogs/ AWSLogs/20230920/ AWSLogs/20230921/ AWSLogs/20230922/ aws-logs-write-test done/ done/AWSLogs/20230920/fuga done/AWSLogs/20230920/hoge done/AWSLogs/20230920/piyo done/AWSLogs/20230921/ done/AWSLogs/20230921/fuga done/AWSLogs/20230921/hoge done/AWSLogs/20230921/piyo done/AWSLogs/20230922/fuga done/AWSLogs/20230922/hoge done/AWSLogs/20230922/piyo exportedlogs/8476bff8-c2da-49b1-ba9d-1fb6af96e2c8/eni-05bca160943029114-all/000000.gz
次にマッチするオブジェクトが0件となるようにプリフィックスを与えて実行してみましょう。
ページイテレータに’Contents’キーがないためキーエラーとなります。
list_objects_v2では条件にマッチするオブジェクトが0件の場合、空の’Contents’を返すのではなく、’Contents’そのものがレスポンスに含まれません。
私自身も0件マッチの場合は空の’Contents’を返すのだろうと思い込んでおり、この状況を目にしたことがあります。
python s3_getObjectKeys_bad.py koha-test invalid-prefix
Traceback (most recent call last):
File "/Users/kohayagawa/Library/CloudStorage/Box-Box/koh/work/clp_cloudpack_section5/cli/blog/s3_getObjectKeys_bad.py", line 32, in
main()
File "/Users/kohayagawa/Library/CloudStorage/Box-Box/koh/work/clp_cloudpack_section5/cli/blog/s3_getObjectKeys_bad.py", line 27, in main
for key in getObjectKeys(bucket, prefix):
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/Users/kohayagawa/Library/CloudStorage/Box-Box/koh/work/clp_cloudpack_section5/cli/blog/s3_getObjectKeys_bad.py", line 12, in getObjectKeys
return [
^
File "/Users/kohayagawa/Library/CloudStorage/Box-Box/koh/work/clp_cloudpack_section5/cli/blog/s3_getObjectKeys_bad.py", line 15, in
for content in page['Contents']
~~~~^^^^^^^^^^^^
KeyError: 'Contents'
ベターな書き方
ではコードを修正してみましょう。
レスポンスに'Contents'キーが含まれるかチェックの必要がありますね。
レスポンスは辞書型となっているため、getメソッドでフィールドにアクセスすることができます。また、キーが存在しない場合のデフォルト値を設定することができるので便利です。（参考）
この方法によるレスポンスフィールドへのアクセスの利点は以下となります。
- キーの存在確認とデフォルト値の設定が一度の呼び出しで行なえます。コンパクトなコードになります。
- キーが存在しない場合のデフォルト値を設定できるため、エラーハンドリングが簡潔です。
- response.get('key', {}).get('subkey', default_value) のようにネストしたアクセスもすっきりと記述できます。
今回のケースではget('Contents', []) と書くことで、'Contents'キーが含まれない場合は空のリストを得ることができるため、リスト内包表記で書いたreturn文を維持することができます。（if文でキーが含まれるかチェックするのは面倒ですよね！）
以下が修正したコードになります。15行目を見て頂ければと思います。
import argparse
from boto3.session import Session
from mypy_boto3_s3 import S3Client
session = Session()
def getObjectKeys(bucket: str, prefix: str) -> list[str]:
client: S3Client = session.client('s3')
paginator = client.get_paginator('list_objects_v2')
return [
content['Key']
for page in paginator.paginate(Bucket=bucket, Prefix=prefix)
for content in page.get('Contents', [])
]
def main():
parser = argparse.ArgumentParser()
parser.add_argument('bucket', help='name of the s3 bucket.')
parser.add_argument('prefix', help='prefix for listing s3 objects.')
args = parser.parse_args()
bucket = args.bucket
prefix = args.prefix
for key in getObjectKeys(bucket, prefix):
print(key)
if __name__ == '__main__':
main()
修正したコードでは、マッチするオブジェクトが0件でもキーエラーは発生しません。
$ python s3_getObjectKeys_good.py koha-test invalid-prefix （出力なし）
その他のリソースは？
上記ではS3バケットのオブジェクト取得を例に'Contents'キーの有無に着目してコードを修正してみました。他のリソースでも同様の考慮は必要なのか気になったので、いくつか試してみました。
EC2
インスタンスを配置していないアイルランドリージョンでdescribe_instancesを試しました。
空のReservationsが返されています。
from boto3.session import Session
from mypy_boto3_ec2.client import EC2Client
session = Session()
client: EC2Client = session.client('ec2', region_name='eu-west-1')
paginator = client.get_paginator('describe_instances')
for page in paginator.paginate():
for resavation in page.get('Reservations', []):
print('hoge')
print(resavation.get('Instances', 'Nothing'))
$ python ec2_describeInstances.py （出力なし）
SSM
パラメータを作成していないアイルランドリージョンでdescribe_parametersを試しました。
こちらも空のParametersが返されています。
from boto3.session import Session
from mypy_boto3_ssm.client import SSMClient
session = Session()
client: SSMClient = session.client('ssm', region_name='eu-west-1')
paginator = client.get_paginator('describe_parameters')
for page in paginator.paginate():
print(page.get('Parameters', 'Nothing'))
$ python ssm_describeParameters.py []
RDS
RDSのないアイルランドリージョンでのdescribe_db_instancesを試しました。
こちらも空のDBInstancesが返されるようです。
from boto3.session import Session
from mypy_boto3_rds.client import RDSClient
session = Session()
client: RDSClient = session.client('rds', region_name='eu-west-1')
paginator = client.get_paginator('describe_db_instances')
for page in paginator.paginate():
print(page.get('DBInstances', 'Nothing'))
$ python rds_describeDbInstances.py []
調査結果
簡易な調査ではありますが、S3と同様にキーが含まれないパターンは見つけることができませんでした・・・。
同様のパターンを発見できれば、get()メソッドを使おうねと楽に主張できたのですが・・・。
まとめ
S3バケット中のオブジェクト一覧の取得というケースで、レスポンス中に’Contents’が含まれないことがあること、そのエラーハンドリングとして辞書型のget()メソッドの利用を紹介させていただきました。
他のリソースについても同様にキーが含まれないことがあるのか（簡易に）調査してみたところ、同様のパターンは見つけることはできずとなりました。
しかしながら、boto3に限らず他のサービスのAPIからのレスポンスを利用の際には、get()メソッドを利用することで簡潔なコードでエラーハンドリングを行うことができそうです。
ここまで読んでいただきありがとうございました。アサンテ！
