Python으로 배치 IP 지오로케이션 처리하기
POST /v1/lookup을 사용해 IP 목록을 100개씩 배치로 보강 처리하며, 행 연결, 체크포인트 출력, 429 백오프를 함께 다룹니다 — 공식 Python SDK가 이 엔드포인트를 지원하지 않으므로 HTTP를 직접 호출합니다.
시작하기 전에
- Watch-IP 계정과 시크릿 조회 키 — POST /v1/lookup은 서버 간 전용이므로 브라우저에서 호출하거나 클라이언트에 키를 담아 배포하지 마세요.
- Python 3.9 이상.
- 보강 처리할 IP의 로컬 목록 — 이 가이드에서는 한 줄에 하나씩 IP가 담긴 CSV 파일에서 읽어옵니다.
1. SDK 대신 HTTP를 직접 사용하는 이유
공식 Python 패키지인 watch-ip-sdk는 브라우저용 방문자 엔드포인트인 GET /v1/geo만 구현하고 있으며, POST /v1/lookup에 대한 메서드는 없습니다. 이 가이드에서는 새 의존성을 추가하는 대신 SDK와 동일한 무의존성 방식을 따라, 표준 라이브러리의 urllib로 API를 직접 호출합니다.
2. 입력 목록 읽고 검증하기
각 IP의 원래 행 위치를 유지해, 결과를 원본 레코드와 다시 매칭할 수 있도록 하세요 — API는 전달받은 순서와 동일한 순서로 결과를 반환하며, 이는 자체 목록의 순서도 그대로 유지될 때만 도움이 됩니다.
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~100개의 IP를 받으며, 조회를 시작하기 전에 배치 전체를 검증합니다 — 잘못된 형식의 주소가 하나라도 있으면 청크 전체가 실패하므로, 입력 목록이 지저분할 가능성이 있다면 먼저 로컬에서 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. 중단된 작업 재개하기
결과는 배치 단위로 하나씩 추가되므로, 충돌 후 재시작할 때는 목록 전체를 다시 제출할 필요 없이 이미 results.jsonl에 기록된 IP만 건너뛰면 됩니다 — 배치 내 각 입력은 중복 제거 없이 개별적으로 과금되므로, 중복된 IP를 다시 제출하면 다시 과금됩니다.
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 조회 레퍼런스를, 이 엔드포인트가 반환할 수 있는 전체 상태 코드 목록은 오류 및 속도 제한 문서를 참고하세요.