Amazon SP-API の Report API を用いてレポートを取得する

Author:suzu

はじめに

どうも、こんにちは。Belong で Software Engineer として働いている suzu です。
業務で Amazon の在庫データを扱う機会があり、Amazon SP-API（Selling Partner API）の Report API を利用しました。

Report API は「リクエストしたらすぐデータが返ってくる」タイプの API ではなく、Amazon 側でレポートを非同期に生成し、完了後にダウンロードするという少し独特な流れを持っています。初見だと戸惑いやすいポイントなので、この記事では

• Report API の全体像
• レポートをオンデマンドで取得する手順（3 ステップ）
• 生成済みレポートの一覧取得（getReports）と、定期実行の概要

を、実際に動く Python のコードとともに整理します。

なお、本記事のサンプルは Amazon 公式の Python SDK である amzn-sp-api¹ を利用しています。API 仕様の一次情報は公式リファレンス² を参照してください。

事前準備

認証情報の用意

SP-API は LWA（Login with Amazon）による OAuth2 で認証します。事前に以下の 3 つを取得しておきます。取得手順は公式ガイド³を参照してください。

• client_id
• client_secret
• refresh_token

SDK の導入

本記事ではパッケージ管理に Poetry を想定し、次のコマンドで依存に追加します。

poetry add amzn-sp-api

pip を使っている場合は pip install amzn-sp-api に読み替えてください。いずれの場合も、追加すると spapi という名前でパッケージが利用できます。

リージョンとマーケットプレイス

SP-API はリージョンごとにエンドポイントが分かれています。SDK では region に以下のいずれかを指定します。

日本のマーケットプレイスを対象にする場合は、region="FE" と marketplaceIds=["A1VC38T7YXB528"]（日本の marketplace ID）を指定します。

Report API の全体像

Report API のオペレーションは、大きく 3 つのグループに分かれます。

ポイントは、レポートは即座に手に入らないという点です。createReport を呼ぶと Amazon 側で生成処理がキューイングされ、完了までに時間がかかります。そのため、

1. createReport で生成をリクエストし、reportId を受け取る
2. getReport を一定間隔で問い合わせ（ポーリング）、processingStatus が DONE になって reportDocumentId が付与されるのを待つ
3. getReportDocument で署名付き URL を取得し、その URL からデータをダウンロードする

という 3 ステップが基本になります。まずは、オンデマンド取得で実践してみましょう。

オンデマンドでレポートを取得する

ここではFBA在庫の情報取得に利用できる GET_FBA_INVENTORY_PLANNING_DATA レポートを例に、3 ステップを実装します。

クライアントの初期化

SPAPIConfig に認証情報とリージョンを渡し、SPAPIClient を経由して ReportsApi を組み立てます。

from spapi.auth.credentials import SPAPIConfig
from spapi.client import SPAPIClient
from spapi.api.reports_v2021_06_30.reports_api import ReportsApi


def build_reports_api(client_id, client_secret, refresh_token, region="FE"):
    config = SPAPIConfig(
        client_id=client_id,
        client_secret=client_secret,
        refresh_token=refresh_token,
        region=region,  # NA, EU, FE, SANDBOX
        scope=None,
    )
    client = SPAPIClient(config)
    return ReportsApi(client.api_client)

ステップ 1: レポートの生成をリクエストする（createReport）

create_report に、レポート種別・対象期間・マーケットプレイスを指定します。返ってくるのは reportId だけで、この時点ではまだデータはありません。

import datetime

reports_api = build_reports_api(client_id, client_secret, refresh_token)

one_month_ago = datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=31)
payload = {
    "reportType": "GET_FBA_INVENTORY_PLANNING_DATA",
    "dataStartTime": one_month_ago.strftime("%Y-%m-%dT00:00:00Z"),
    "marketplaceIds": ["A1VC38T7YXB528"],  # 日本
}

create_response = reports_api.create_report(payload)
report_id = create_response.report_id

dataStartTime / dataEndTime は ISO 8601 形式で、レポートが対象とするデータの期間を表します（レポート種別によっては不要です）。

ステップ 2: 生成完了を待つ（getReport をポーリング）

create_report の直後は、まだ reportDocumentId が付与されていません。そこで get_report を一定間隔で呼び、reportDocumentId が返ってくるまで待ちます。

processingStatus は次の値を取ります。DONE になったときに reportDocumentId が付きます。

• IN_QUEUE … 生成待ち
• IN_PROGRESS … 生成中
• DONE … 完了（reportDocumentId が付与される）
• CANCELLED … キャンセルされた
• FATAL … 生成に失敗した

from time import sleep


def wait_for_document_id(reports_api, report_id, tries=5, interval=10):
    for i in range(tries):
        report = reports_api.get_report(report_id)
        document_id = report.report_document_id
        if document_id is not None:
            return document_id
        # 生成中のため少し待って再試行
        sleep(interval)
    raise RuntimeError(f"{tries} 回試行しましたがレポートが生成されませんでした")


document_id = wait_for_document_id(reports_api, report_id)

リトライ回数や間隔はレポートのサイズや種別によって調整が必要です。大きなレポートは数分かかることもあります。本番運用では processingStatus が FATAL / CANCELLED になったケースも明示的にハンドリングすると安全です。

ステップ 3: レポートをダウンロードする（getReportDocument）

get_report_document は、実データそのものではなく 署名付き URL（presigned URL） を返します。この URL に対して HTTP GET を行うことで、実データを取得します。

import io
import requests

document = reports_api.get_report_document(
    document_id,
    enable_content_encoding_url_header=True,
)

response = requests.get(document.url, timeout=30)
response.raise_for_status()
data = io.StringIO(response.content.decode("utf-8"))

今回の GET_FBA_INVENTORY_PLANNING_DATA はタブ区切り（TSV）で返ってくるので、そのまま pandas で読み込めます。

import pandas as pd

df = pd.read_csv(data, delimiter="\t", encoding="utf-8")
print(df.head())

※ フォーマットは種別により異なり、TSV のほか JSON や XML で返るものもあります。

これでオンデマンド取得は完了です。createReport → getReport（ポーリング）→ getReportDocument の 3 ステップさえ押さえれば、reportType を変えるだけで他のレポートにも応用できます。

署名付き URL の注意点

• 有効期限が短い（数分程度）ため、取得したらすぐにダウンロードします。
• レポートによっては compressionAlgorithm に GZIP が指定されて返ることがあります。enable_content_encoding_url_header=True を指定するとクライアントが対応していれば自動で展開されますが、そうでない場合はダウンロード後に明示的な解凍が必要です。
  • 筆者の環境では 1000KB を少し超えたあたりで圧縮がかかりました。データ量で挙動が変わるので、最初から enable_content_encoding_url_header=True を指定しておくと安心です。

生成済みレポートを一覧取得する（getReports）

生成済みのレポートは get_reports（レポート一覧の取得）で拾えます。reportTypes や processingStatuses、createdSince で絞り込めるので、「対象種別で、生成完了（DONE）のもの」を後から取得できます。

import datetime

reports = reports_api.get_reports(
    report_types=["GET_FBA_INVENTORY_PLANNING_DATA"],
    processing_statuses=["DONE"],
    created_since=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1),
)

for report in reports.reports:
    document_id = report.report_document_id
    # あとはオンデマンド取得のステップ 3 と同じ流れでダウンロードできる

得られた reportDocumentId を使えば、ステップ 3（getReportDocument）と同じ流れでダウンロードできます。

（補足）定期実行：Report Schedule

createReport を都度呼ぶ代わりに、Amazon 側にスケジュールを登録して定期生成させる Report Schedule（createReportSchedule など）もあります。ただし対応する種別は一部に限られます。スケジュールで生成されたレポートの取得は、前述の getReports で拾う方法と、Notifications API⁴（AWS 環境が前提）で完了通知を受け取る方法のいずれでも行えます。

つまずきポイント

最後に、実装時にハマりやすい点をまとめておきます。

• 生成完了までの待ち時間: createReport 直後は reportDocumentId が付いていません。必ずポーリングで待つ必要があります。
• 圧縮形式: レポートによっては GZIP で返ります。compressionAlgorithm を確認して解凍処理を用意します。
• レートリミット: 各オペレーションにはレートリミットが設定されています。特に createReport / getReportDocument は低頻度なので、短時間に何度も呼ばない設計にします（最新の値は公式リファレンス² を参照）。

まとめ

Amazon SP-API の Report API は、

• createReport → getReport（ポーリング）→ getReportDocument の 3 ステップが基本
• reportType を変えるだけであらゆるレポートに応用できる
• 定期取得は Report Schedule でも自動生成できる（ただし対応する種別は一部に限られる）

という構造になっています。「非同期でレポートを生成・取得する」というモデルさえ理解すれば、あとはレポート種別を選ぶだけです。指定できる reportType の一覧や各オペレーションの詳細は公式リファレンス² にまとまっているので、あわせて参照してみてください。

弊社 Belong では一緒に働く仲間を募集しています。

Python を使った課題解決や、Go でのサーバーサイド開発に興味のある方など、
エンジニアリングチーム紹介ページをご覧いただけると幸いです。