概要

stripeの支払データを月別で集計する必要があったので、調査して対応しました。
backendにはpythonを使っていたので、pythonのライブラリを使って対応しました。

以下のようにして対応しました。

• list invoices APIを使って、特定の期間の支払済データを全て取得する
• apiからデータ取得後、データをサマリする

実装

以下のような感じになりました。

ポイントとしては以下のような感じです。

• invoiceリスト取得のクエリパラメータに使うtimestampを生成（timezoneを意識する必要あり）
• メモリなどのリソースが枯渇しないように、APIの一回あたりの取得件数の上限を決める
• apiはcreatedの降順に並んで返却されるため一括で取得できなかった場合は、最後のinvoice idを使って、それ以降のものを取得する

import stripe
from datetime import datetime, date
import time
from pytz import timezone
from dateutil.relativedelta import relativedelta

stripe.api_key = "your_stipe_secret_key_starting_with_sk_here"

# 当月と次月の初日のtimestampをcreatedパラメータに指定する
from_date = date(2020, 9, 1)
from_datetime = datetime.combine(from_date, datetime.min.time())
from_datetime = timezone('Asia/Tokyo').localize(from_datetime)
from_time = int(from_datetime.timestamp())
end_date = date(2020, 10, 1)
end_datetime = datetime.combine(end_date, datetime.min.time())
end_datetime = timezone('Asia/Tokyo').localize(end_datetime)
end_time = int(end_datetime.timestamp())

created = {
  "gte": from_time,
  "lt": end_time,
}
limit=1000
last_invoice_id = None


while True:
    api_result = stripe.Invoice.list(status="paid", created=created, ending_before=last_invoice_id, limit=limit)
    invoices = api_result["data"]
    for invoice in invoices:
        last_invoice_id = invoice["id"]
        # ここで集計処理
    
    if len(invoices) < limit:
        break

その他考慮すること

webhookなどを使って、支払があった度に支払のあったデータを集計することで、ほぼリアルタイムに集計することができますが
webhookはapplicationエラーやメンテナンス時の考慮などで、データが抜け落ちる可能性があるため、
一度で集計できるようにしています。

もしリアルタイムに集計したいというニーズがある場合は、以下のようにすると良いかなと考えています。

1. webhookを使って、支払が成功する度に集計処理を実行する
2. また集計処理をするタイミングで、前回集計が完了したidをdbなどに保存しておき、webhookでやってきた場合に前回成功したID以降のものを取得して再集計する
3. （optional）例えば、1日おきくらいで、集計が漏れている支払データがないかcronなどでチェックしてあれば、取得して再集計する

このようにすることで、処理が失敗した場合や、webhookを受け取れなかった場合の考慮もできるかなと思います。

以上です。