AWS Cost Explorer API のページネーションをがんばる

はじめに

数年前に以下のような記事を投稿しました。

AWS Cost Explorer API を使用してアカウントごとの日次コストを取得するという内容ですが、取得するアカウント数の増加や日数によってはページ分割が発生します。

この記事の中で使用している get_cost_and_usage メソッドは Boto3 の Pagenator を利用できません。また API の応答に TimePeriod が含まれており、重複しない完全な応答を得るのに一工夫必要だったため、概要を紹介します。

Pagenator を使えばいいのでは？

Boto3 では Pagenator によりページネーション処理を抽象化し、すべての結果を簡単に取得することができます。

しかし、サービスや API によって Pagenator のサポート状況が異なり、Cost Explorer では使用できません。

以下のような簡易的な Lambda 関数のコードを例にして考えます。

import datetime
import boto3

def lambda_handler(event, context):
    today = datetime.date.today()
    start = today.replace(day=1).strftime('%Y-%m-%d')
    end = today.strftime('%Y-%m-%d')

    ce = boto3.client('ce')
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start,
            'End' :  end,
        },
        Granularity='DAILY',
        Metrics=[
            'NetUnblendedCost'
        ],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'LINKED_ACCOUNT'
            }
        ]
    )
    return response['ResultsByTime']

本来であれば以下のように書きたいのですが、

import datetime
import boto3

def lambda_handler(event, context):
    today = datetime.date.today()
    start = today.replace(day=1).strftime('%Y-%m-%d')
    end = today.strftime('%Y-%m-%d')

    ce = boto3.client('ce')
    paginator = ce.get_paginator('get_cost_and_usage')
    page_iterator = paginator.paginate(
        TimePeriod={
            'Start': start,
            'End' :  end,
        },
        Granularity='DAILY',
        Metrics=[
            'NetUnblendedCost'
        ]
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'LINKED_ACCOUNT'
            }
        ]
    )
    response = []
    for page in page_iterator:
        response.extend(page['ResultsByTime'])
    return response

Pagenator がサポートされていないサービスやメソッドではエラーが発生します。

"errorMessage": "Operation cannot be paginated: get_cost_and_usage"
"errorType": "OperationNotPageableError"

自力でかんばる

多くの記事で紹介されているように、API レスポンスに含まれる NexToken や NextPageToken の有無をチェックして再帰処理を行うことで結果セット全体を取得できます。

get_cost_and_usage メソッドのレスポンスには、ある期間のコストを示すために TimePeriod が含まれています。

例えば 2025-01-15 ～ 2025-01-16 期間のコストを出力する途中でページ分割された場合、次のページにも同じ TimePeriod が含まれることになります。

# 1 ページ目の応答の最後
{
    "TimePeriod": {
        "Start": "2025-01-15",
        "End": "2023-01-16"
    },
    "Total": {},
    "Groups": [
        {
            "Keys": ["123456789012"],
            "Metrics": {...}
        }
    ],
    "Estimated": false
}

# 2 ページ目の応答の最初
{
    "TimePeriod": {
        "Start": "2023-01-15",  # 同じ期間
        "End": "2023-01-16"
    },
    "Total": {},
    "Groups": [
        {
            "Keys": ["987654321098"],  # 異なるアカウントID
            "Metrics": {...}
        }
    ],
    "Estimated": false
}

そのため、分割取得した結果を単純に結合するだけでは、結果を後続の処理で利用する際に不便です。重複する期間がある場合は以下のように Groups のリストのみを結合したくなります。

{
    "TimePeriod": {
        "Start": "2025-01-15",
        "End": "2023-01-16"
    },
    "Total": {},
    "Groups": [
        {
            "Keys": ["123456789012"],
            "Metrics": {...}
        },
        {
            "Keys": ["987654321098"],
            "Metrics": {...}
        }
    ],
    "Estimated": false
}

前置きが長くなりましたが、以下のようなコードでページ分割された場合でも同じ期間のデータを適切にマージできます。

import datetime
import boto3

def lambda_handler(event, context):
    today = datetime.date.today()
    start = today.replace(day=1).strftime('%Y-%m-%d')
    end = today.strftime('%Y-%m-%d')

    ce = boto3.client('ce')
    all_results = []
    next_token = None
    seen_periods = set() 
    
    while True:
        params = {
            'TimePeriod': {
                'Start': start,
                'End': end
            },
            'Granularity': 'DAILY',
            'Metrics': ['NetUnblendedCost'],
            'GroupBy': [
                {
                    'Type': 'DIMENSION',
                    'Key': 'LINKED_ACCOUNT'
                }
            ]
        }
        if next_token:
            params['NextPageToken'] = next_token
        
        response = ce.get_cost_and_usage(**params)

        # レスポンスの重複チェックと結合処理
        for result in response['ResultsByTime']:
            # 期間を一意に識別するためのキーとして開始日と終了日のタプルを作成
            # 例: ('2024-12-01', '2024-12-02')
            period_key = (result['TimePeriod']['Start'], result['TimePeriod']['End'])

            # seen_periods セットに期間キーが存在するかチェック
            if period_key in seen_periods:
                # 対象の期間のデータを検索し Groups を結合
                existing_result = next(r for r in all_results
                                    if r['TimePeriod']['Start'] == period_key[0]
                                    and r['TimePeriod']['End'] == period_key[1])
                existing_result['Groups'].extend(result['Groups'])
            else:
                # 新しい期間のデータはそのままリストに追加
                all_results.append(result)
                # 処理済み期間のセットに期間キーを追加
                seen_periods.add(period_key)

        # 次のページがない場合は処理を終了
        next_token = response.get('NextPageToken', None)
        if not next_token:
            break

    return all_results

もしもっと簡単な方法があったら教えてください。

以上です。
参考になれば幸いです。