Python 강좌 / 종합 실습: API 데이터 수집하고 중복 없이 저장하기

Python 강좌

API 수집기는 응답 전체를 검증한 뒤 외부 ID를 기준으로 저장하면 재실행 시 중복을 줄일 수 있습니다. 이 실습은 같은 ID의 최신 이름·금액을 갱신하는 정책으로 JSON 요청과 SQLite 트랜잭션을 연결합니다.

식별자와 재실행 정책

외부 id가 같은 거래를 가리킨다는 계약이 있어야 upsert가 의미가 있습니다. 금액과 이름이 같다는 이유로 다른 거래를 하나로 합치지 않습니다. 예제는 한 응답 안에 같은 id가 있으면 뒤의 값이 최종 값이 되는 정책이며, 충돌을 오류로 처리해야 한다면 저장 전 중복 검사로 바꿉니다.

관련 동작과 지원 조건은 종합 실습: API 데이터 수집하고 중복 없이 저장하기 관련 공식 문서에서 확인할 수 있습니다.

모든 행을 먼저 검증하여 잘못된 입력이 한 건이라도 있으면 저장을 시작하지 않습니다. DB 쓰기는 하나의 트랜잭션으로 처리하지만 네트워크 요청과 DB 작업 전체가 분산 트랜잭션이 되는 것은 아닙니다.

로컬 응답 준비와 실행

expenses.json 파일의 전체 내용입니다.

[{"id":"e1","name":"coffee","amount":3500},{"id":"e2","name":"lunch","amount":8000}]

첫 번째 터미널에서 테스트 서버를 시작합니다.

python -m http.server 8000 --bind 127.0.0.1

두 번째 터미널에서 아래 파일을 저장하고 실행합니다. 작업 폴더는 다른 실습과 분리합니다.

전체 수집 프로그램

collect.py 파일의 전체 내용입니다.

import json
import sqlite3
import sys
from urllib.request import urlopen

with urlopen(sys.argv[1], timeout=5) as response:
    if response.status != 200:
        raise ValueError("unexpected status")
    raw = response.read(1_000_001)
    if len(raw) > 1_000_000:
        raise ValueError("response too large")
    data = json.loads(raw.decode("utf-8"))
if not isinstance(data, list) or len(data) > 10000:
    raise ValueError("invalid batch")
records = []
for row in data:
    if not isinstance(row, dict):
        raise ValueError("invalid row")
    key, name, amount = row.get("id"), row.get("name"), row.get("amount")
    if not isinstance(key, str) or not key.strip() or not isinstance(name, str) or not name.strip():
        raise ValueError("invalid identity")
    if type(amount) is not int or amount < 0:
        raise ValueError("invalid amount")
    records.append((key, name, amount))
con = sqlite3.connect("collected.db", autocommit=False)
try:
    with con:
        con.execute("CREATE TABLE IF NOT EXISTS expenses (id TEXT PRIMARY KEY, name TEXT NOT NULL, amount INTEGER NOT NULL CHECK(amount >= 0))")
        con.executemany("INSERT INTO expenses VALUES (?, ?, ?) ON CONFLICT(id) DO UPDATE SET name=excluded.name, amount=excluded.amount", records)
    count, total = con.execute("SELECT COUNT(*), COALESCE(SUM(amount),0) FROM expenses").fetchone()
    print("stored:", count, "total:", total)
finally:
    con.close()

python collect.py http://127.0.0.1:8000/expenses.json

stored: 2 total: 11500을 출력합니다. 같은 응답으로 다시 실행해도 4건으로 늘지 않고 2건을 유지해야 합니다. 수집 중 실패하면 traceback과 종료 코드를 확인하고 후속 처리를 진행하지 않습니다.

실패·갱신·완료 확인

같은 응답을 두 번 처리해 행 수와 합계를 비교합니다.
e2의 금액만 9000으로 바꿔 2건·12500으로 갱신되는지 확인합니다.
금액을 문자열로 바꿔 검증 실패 뒤 기존 DB가 유지되는지 확인합니다.
서버를 Ctrl+C로 종료합니다.

외부 서비스에서는 페이지 나누기, 요청 제한, 인증과 변경·삭제 정책도 필요합니다. 일부 페이지만 성공한 상태를 전체 수집 완료로 기록하지 않습니다. 무제한 재시도를 넣거나 모든 실패를 빈 리스트로 바꾸면 데이터 누락을 숨기므로 실패 상태를 명확히 전달합니다.

Python 강좌