List系APIの取得数上限とPaginator

はじめに

「ファイル一覧、job一覧などを取得して何かしたい」というのはちょっとしたスクリプトからシステム開発までよく出くわす場面です。
そして、そういったもののためにweb serviceにはList系APIが実装されていることが一般的です。

そのようなList系APIはレスポンスが巨大になりがちなため、 取得数に上限が設けられることが一般的 であり、 この取得上限に注意しながら実装を進める必要があります。
その一方で、ライブラリによってはPaginatorという、 取得数上限を考慮せずに済むような仕組み が準備されているものがあります。

本記事では、aws s3 ListObjectsV2 API を例にとり、 取得数上限を考慮した実装と、Paginatorの説明をします。

推奨の方法をいち早く確認したい場合は、 Paginatorを利用する方法 を参照ください。
また、本記事では python + boto3 を利用します。

List APIの仕様と、取得数上限を考慮したコーディング

aws s3 ListObjectsV2 API の説明にはこのような記載があります：

Returns some or all (up to 1,000) of the objects in a bucket with each request.
（バケット内のobjectの一部またはすべて（1,000個まで）を返します。）

このようなページ分割された（Paginated）情報を送ってくるAPIを利用してすべてのobjectを取得するには何らかの工夫が必要です。
そして、勿論この為の仕組みが準備されており、URI Request ParametersとResponse Elementsに以下のような記述を発見します：

URI Request Parameters::continuation-token
（意訳： ContinuationToken は、トークンを利用することでリストがまだ続いていることをAmazon S3に示します。
ContinuationToken は難読化されており、実際のキーではありません。）

Response Elements::NextContinuationToken
（意訳： NextContinuationToken は、Response Elementsである isTruncated が true のときに送られます。これは、bucketにまだリストできるobjectが存在することを示しています。
Amazon S3への次のリストのリクエストは、この NextContinuationToken で続行できます。
NextContinuationToken は難読化されており、実際のキーではありません。）

...というわけで、 NextContinuationToken が得られた場合は再度 continuation-token に NextContinuationToken の値を仕込んでリクエストを投げればよい、ということになります。

code

import boto3

boto3.setup_default_session(profile_name='default')
client = boto3.client('s3')
bucket = 'some-bucket-name'

def paginate_list_objects_v2(**kwargs):
    payload = kwargs
    while 1:
        response = client.list_objects_v2(**payload)
        yield response
        if not response.get('IsTruncated'):
            return
        payload['ContinuationToken'] = response['NextContinuationToken']

payload = {
    'Bucket': bucket,
}
for response in paginate_list_objects_v2(**payload):
    print(response.get('Contents'))

Paginatorを利用する方法

上記のようなコードを書けばすべてのobjectが取得できますが、面倒と言えば面倒ですよね。
そこで、boto3にはPaginatorと呼ばれる仕組みが存在しています。
ページ分割周りを全部巻き取ってくれるため、コードがとても簡潔になります。

code

import boto3

boto3.setup_default_session(profile_name='default')
client = boto3.client('s3')
bucket = 'some-bucket-name'

payload = {
    'Bucket': bucket,
}
for response in client.get_paginator('list_objects_v2').paginate(**payload):
    print(response.get('Contents'))

ContinuationToken 周りを考慮せずに済み、とても簡潔に書くことができました。

まとめ

ページ分割された（Paginated）情報を送ってくるAPIを利用してすべての情報を取得する方法を、コードベタ書きとPaginatorを利用した場合との2種類で説明しました。
Paginatorを利用すれば簡潔に書けるためおすすめですが、必ずしもライブラリにPaginatorが存在するとは限らないため、いつでもベタ書きで書けるように準備しておきましょう。