GUIDES

バッチ処理のしかた

大量の住所を非同期で処理する手順です。送信・ポーリング・結果の取得までを、Python のコードで示します。Growth 以上で使えます。

更新: 2026年10月12日

数百件以上の住所は、1件ずつ呼ぶより、バッチにまとめるほうが簡単です。Growth 以上で使えます(Developer・Starter は 403、課金なし)。

流れ

  1. 住所を、プランの上限(Growth 100・Business 2,000・Enterprise 50,000)以下に分けて送る。
  2. 202 と、結果を取りに行く URL を受け取る。
  3. その URL を、間隔をあけて取りに行く。
  4. 終わったら、結果を取り出す。1時間で破棄されます。

コード(Python)

import os, time, requests

KEY = os.environ["KITEN_API_KEY"]
BASE = "https://api.zorynto.com"   # TODO(要確認): API ホスト
H = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}

def chunks(xs, n):
    for i in range(0, len(xs), n):
        yield xs[i:i + n]

def run_batch(addresses):
    r = requests.post(f"{BASE}/v1/batch", headers=H, json={"addresses": addresses}, timeout=30)
    r.raise_for_status()                       # 202
    status_url = r.json()["url"]               # TODO(要確認): 応答の項目名
    wait = 1
    while True:
        s = requests.get(status_url, headers=H, timeout=30)
        if s.status_code == 429:
            time.sleep(float(s.headers.get("Retry-After", 1)))
            continue
        s.raise_for_status()
        body = s.json()
        if body.get("done"):                   # TODO(要確認): 完了の判定
            return body["results"]
        time.sleep(wait)
        wait = min(wait * 2, 10)

results = []
for part in chunks(all_addresses, 100):        # Growth の場合
    results += run_batch(part)

要確認 応答の項目名(url done results)は、公開前に確定します。上のコードは、流れを示すための例です。

注意

  • 課金は1件ずつです。 受付時に、月間の残りで足りないと拒否され、先に引かれた分は戻されます。
  • 結果を取れるのは、バッチを作ったキーだけです。
  • 取りに行く間隔は、1〜2秒から始め、少しずつ延ばします。429 が返ったら、Retry-After に従います。
  • 結果は1時間で消えます。取得したら、すぐに自社の側へ保存してください。(保持の延長は、Business 向けに計画中です。)

関連:GET /v1/batch/{id}・再試行と429

この内容に誤りや不足があれば、お知らせください。