GUIDES
バッチ処理のしかた
大量の住所を非同期で処理する手順です。送信・ポーリング・結果の取得までを、Python のコードで示します。Growth 以上で使えます。
更新: 2026年10月12日
数百件以上の住所は、1件ずつ呼ぶより、バッチにまとめるほうが簡単です。Growth 以上で使えます(Developer・Starter は 403、課金なし)。
流れ
- 住所を、プランの上限(Growth 100・Business 2,000・Enterprise 50,000)以下に分けて送る。
- 202 と、結果を取りに行く URL を受け取る。
- その URL を、間隔をあけて取りに行く。
- 終わったら、結果を取り出す。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)
要確認 応答の項目名(
urldoneresults)は、公開前に確定します。上のコードは、流れを示すための例です。
注意
- 課金は1件ずつです。 受付時に、月間の残りで足りないと拒否され、先に引かれた分は戻されます。
- 結果を取れるのは、バッチを作ったキーだけです。
- 取りに行く間隔は、1〜2秒から始め、少しずつ延ばします。429 が返ったら、
Retry-Afterに従います。 - 結果は1時間で消えます。取得したら、すぐに自社の側へ保存してください。(保持の延長は、Business 向けに計画中です。)
この内容に誤りや不足があれば、お知らせください。