PythonによるバッチIPジオロケーション
行の対応関係の保持、チェックポイント付き出力、429時のバックオフを備えて、100件ずつのバッチでPOST /v1/lookupに対してIPのリストをエンリッチします—公式のPython SDKはこのエンドポイントをカバーしていないため、HTTPを直接呼び出します。
始める前に
- Watch-IPアカウントとシークレットルックアップキー—POST /v1/lookupはサーバー間通信専用です。ブラウザから呼び出したり、クライアントにキーを含めて配布したりしないでください。
- Python 3.9以降。
- エンリッチ対象のIPのローカルリスト—このガイドでは1行に1つのIPを記載したCSVから読み込みます。
1. SDKではなく直接HTTPを使う理由
公式のPythonパッケージであるwatch-ip-sdkは、ブラウザ向けの訪問者用エンドポイントであるGET /v1/geoのみを実装しており、POST /v1/lookup用のメソッドはありません。このガイドでは、SDK自体の依存ゼロという方針に合わせて、このためだけに新しい依存を追加するのではなく、標準ライブラリのurllibでAPIを直接呼び出します。
2. 入力リストを読み込んで検証する
各IPの元の行位置を保持しておくと、結果を元のレコードに突き合わせられます—APIはIPを受け取った順序と同じ順序で結果を返しますが、それが役立つのは自分のリスト自体もその順序を保っている場合だけです。
Python
import csv
with open("ips.csv", newline="") as f:
ips = [row[0].strip() for row in csv.reader(f) if row and row[0].strip()]
print(f"Loaded {len(ips)} IPs")3. 100件ずつのバッチに分割する
POST /v1/lookupは1回の呼び出しにつき1〜100件のIPを受け付け、ルックアップを実行する前にバッチ全体を検証します—不正な形式のアドレスが1件でもあるとチャンク全体が失敗するため、入力リストが汚れている可能性がある場合は先にローカルでIPの構文を検証してください。
Python
def chunk(items, size=100):
for i in range(0, len(items), size):
yield items[i : i + size]4. 各バッチを送信し、結果をチェックポイントする
すべてをメモリに保持して最後にまとめて書き込むのではなく、各バッチが完了するたびに結果を書き込むことで、長いリストの途中で失敗しても既に成功したバッチを失わずに済みます。found: falseは、そのIPの構文は有効だがデータベースにエントリがないことを意味します—これはエラーではなく通常の結果です。
Python
import json
import time
import urllib.error
import urllib.request
API_URL = "https://api.watch-ip.com/v1/lookup"
SECRET_KEY = "YOUR_SECRET_KEY"
def lookup_batch(ips):
request = urllib.request.Request(
API_URL,
data=json.dumps({"ips": ips}).encode(),
headers={
"Authorization": f"Bearer {SECRET_KEY}",
"Content-Type": "application/json",
},
)
with urllib.request.urlopen(request) as response:
return json.load(response)["results"]
with open("results.jsonl", "a") as out:
for batch in chunk(ips):
try:
results = lookup_batch(batch)
except urllib.error.HTTPError as err:
if err.code == 429:
time.sleep(60) # 現在の1分間ウィンドウが終わるまで待機する
results = lookup_batch(batch)
else:
raise
for result in results:
out.write(json.dumps(result) + "\n")5. 中断したジョブを再開する
結果は1バッチずつ追記されるため、クラッシュ後の再開時にはリスト全体を再送信するのではなく、既にresults.jsonlに書き込まれたIPをスキップするだけで済みます—バッチ内の各入力は重複排除されず個別に課金されるため、同じIPを2回送信すると2回分課金されます。
Python
done = set()
with open("results.jsonl") as f:
for line in f:
done.add(json.loads(line)["ip"])
remaining = [ip for ip in ips if ip not in done]エラーとエッジケース
400は、バッチに100件を超えるIPが含まれていた、リストが空だった、またはどこかに不正な形式のアドレスがあったことを意味します—不正なエントリだけでなくチャンク全体が失敗します。403は公開可能キーがシークレットキーの代わりに使われたことを意味します。同じバッチ内にIPv4とIPv6アドレスが混在していても問題ありません—それぞれ独立してルックアップされ、どちらもnullになりうる同じフィールドを持ちます。
次のステップ
正確なレスポンススキーマについてはIP一括ルックアップのリファレンスを、このエンドポイントが返しうるステータスコードの全一覧についてはエラーとレート制限をご覧ください。