こんにちはキーボードの悪魔です。
HHKB Studio良さそうですね。Keychron Q10から浮気しそうです。

今回はBoto3を利用する際のNextTokenの取り扱いについて説明します。
題材としてSSMパラメータストアから指定パスにマッチするパラメータを取得する関数を書いてみましょう。

ダメな例

まずはダメな例というか、素直な例。
以下のコードを準備しました。パスに”/”を指定、つまり全量の取得してその件数をプリントしています。

import boto3
from mypy_boto3_ssm import SSMClient


def getParameters(path: str) -> list:
    client: SSMClient = boto3.client('ssm')
    params = {
        'Path': path
    }
    response = client.get_parameters_by_path(**params)
    parameters: list = response['Parameters']
    return parameters


print(len(getParameters('/')))

実行してみます。10件？
随分とキリの良い数字となりました。
でも、もっとたくさんパラメータあったような？

> python3 code_01.py
10

はい、上記のコードではダメです。残念。。。
全量を取得することを保証できていません。

何がダメなのか？

Responseに含まれていたNextTokenを利用して続きのアイテムを取得できていないことが原因です。

Boto3のドキュメントを読んでみましょう。ssm – get_parameters_by_path

ResponseSyntaxを読み進めるとNextTokenがあります。次のアイテムセットを得るためのトークンだとありますね。

  • NextToken (string) –
    The token for the next set of items to return. Use this token to get the next set of results.

RequestSyntaxにもNextTokenがあります。前回呼び出しのNextTokenをこの引数に割り当てて続きのアイテムを取得することができます。

  • NextToken (string) – A token to start the list. Use this token to get the next set of results.

NextTokenを利用した全量取得

原因は判明しました。でーは、コードを修正してみましょう。

すべきことは・・・

  • レスポンスにNextTokenが含まれるかチェックする
  • 含まれていればNextTokenを利用して続きのアイテムを取得する
  • 上記をNextTokenが含まれないレスポンスを取得するまで繰り返す。

ということですね。再帰関数として記述するのが良さそうです。

修正したコードです。

import boto3
from mypy_boto3_ssm import SSMClient


def getParameters(path: str, nextToken: str = None) -> list:
    client: SSMClient = boto3.client('ssm')
    params = {
        'Path': path
    }
    if nextToken:
        params['NextToken'] = nextToken
    response = client.get_parameters_by_path(**params)
    parameters: list = response['Parameters']
    if 'NextToken' in response:
        parameters.extend(getParameters(path, response['NextToken']))
    return parameters


print(len(getParameters('/')))

実行してみます。23件。
今度こそ全量が取得できたようです。

> python3 code_02.py
23

Paginatorを利用した発展編

NextTokenを意識したコードに修正したことで、パラメータ全量の取得に成功。目的は達しました。
でも、いちいち再帰関数を書くのめんどくさいって思いませんか？
はい、Paginatorの出番です。Paginators

paginatorオブジェクトはget_paginator()関数で作成します。paginatorオブジェクトのpaginate()関数でPageIteratorオブジェクトを取得できます。
というわけで実際にPaginatorを利用したコードが以下になります。リスト内包表記と組み合わせることでかなーりすっきりと書けます。

import boto3
from mypy_boto3_ssm import SSMClient


def getParameters(path: str) -> list:
    client: SSMClient = boto3.client('ssm')
    params = {
        'Path': path
    }
    paginator = client.get_paginator('get_parameters_by_path')
    return [
        parameter
        for page in paginator.paginate(**params)
        for parameter in page['Parameters']
    ]


print(len(getParameters('/')))

まとめ

Boto3を利用する上で必ず直面するNextTokenの取り扱いについて書いてみました。
とりあげたssm – get_parameters_by_path()の他にも、ec2 – describe_instancesやec2 – describe_security_groupsでも必要となるため、意識したコードを書く機会は多いと思われます。s3 – list_object_v2ではContinuationTokenという名前で扱われていたりしますので注意が必要です。都度リファレンスを確認するのが大切ですね。

それではアデュー。