공공데이터 파이썬 수집

공공데이터포털에서 API 키를 받고 나면 다음 관문은 실제로 데이터를 받아오는 일입니다. 브라우저 주소창에 요청 URL을 넣어 결과를 확인하는 것까지는 쉬운데, 그것을 반복 가능한 코드로 만드는 단계에서 손이 멈춥니다.

여기서는 파이썬으로 한 번 호출해 보고, 전체 목록을 받아오고, 파일로 저장하는 데까지 가 보겠습니다. 특정 API에 맞춘 코드가 아니라 대부분의 공공 오픈 API에 통하는 구조로 적었습니다.

준비

필요한 것은 두 가지입니다. 발급받은 인증키, 그리고 쓰려는 API의 기술문서입니다.

기술문서에는 요청 주소와 파라미터 목록이 들어 있습니다. 이름이 조금씩 다르지만 대부분 다음 세 가지는 공통입니다.

• 인증키 — serviceKey인 경우가 많습니다
• 페이지 번호 — pageNo
• 한 페이지당 건수 — numOfRows

응답 형식을 고를 수 있는 API라면 JSON으로 받는 편이 다루기 쉽습니다. 보통 _type 또는 dataType 파라미터로 지정합니다.

한 번 호출해 보기

먼저 응답이 오는지부터 확인합니다.

import requests

SERVICE_KEY = "발급받은_인증키"
URL = "기술문서에_적힌_요청주소"

params = {
    "serviceKey": SERVICE_KEY,
    "pageNo": 1,
    "numOfRows": 10,
    "_type": "json",
}

res = requests.get(URL, params=params, timeout=10)
print(res.status_code)
print(res.text[:500])

여기서 바로 넘어가지 말고 출력된 내용을 확인합니다. 상태 코드가 200인데도 본문에 오류 메시지가 담겨 오는 경우가 흔하기 때문입니다.

자주 만나는 응답은 이렇습니다.

• SERVICE_KEY_IS_NOT_REGISTERED_ERROR — 승인 직후에는 반영까지 시간이 걸립니다. 자동 승인이라도 한 시간 정도 기다려야 하는 경우가 있습니다
• 인증키가 URL 인코딩된 상태로 전달됨 — 포털에서 제공하는 키는 인코딩된 것과 디코딩된 것 두 가지입니다. params에 넣으면 requests가 다시 인코딩하므로 디코딩 키를 씁니다
• XML이 돌아옴 — 그 API가 JSON을 지원하지 않거나 파라미터 이름이 다른 경우입니다

전체를 받아오기

한 번 호출이 성공하면 페이지를 넘겨 가며 전부 가져옵니다. 응답의 totalCount를 보고 몇 페이지인지 계산하는 방식이 안정적입니다.

import math
import time

def fetch_page(page, rows=100):
    params = {
        "serviceKey": SERVICE_KEY,
        "pageNo": page,
        "numOfRows": rows,
        "_type": "json",
    }
    res = requests.get(URL, params=params, timeout=10)
    res.raise_for_status()
    return res.json()

first = fetch_page(1)
body = first["response"]["body"]
total = body["totalCount"]
rows = 100
pages = math.ceil(total / rows)

items = []
for p in range(1, pages + 1):
    data = fetch_page(p, rows)
    chunk = data["response"]["body"]["items"]["item"]
    items.extend(chunk)
    print(f"{p}/{pages} 수집 {len(items)}건")
    time.sleep(0.3)

응답의 구조는 API마다 다릅니다. response - body - items - item 형태가 흔하지만, 항목이 하나뿐일 때 리스트가 아니라 딕셔너리로 오는 API도 있습니다. 처음 한 번은 print로 구조를 확인하고 맞추는 편이 빠릅니다.

time.sleep을 넣은 것은 호출 간격을 두기 위해서입니다. 대부분의 공공 API는 일일 호출 한도가 있고, 짧은 시간에 몰아서 요청하면 차단되는 경우가 있습니다.

저장하기

수집한 결과는 원본 그대로 한 벌 남겨두는 것이 좋습니다. 가공하다 잘못되면 다시 받지 않고 되돌릴 수 있습니다.

import json
import csv
from datetime import date

stamp = date.today().isoformat()

with open(f"raw_{stamp}.json", "w", encoding="utf-8") as f:
    json.dump(items, f, ensure_ascii=False, indent=2)

keys = sorted({k for it in items for k in it.keys()})
with open(f"data_{stamp}.csv", "w", encoding="utf-8-sig", newline="") as f:
    w = csv.DictWriter(f, fieldnames=keys)
    w.writeheader()
    w.writerows(items)

파일명에 날짜를 넣은 것은 기준 시점을 남기기 위해서입니다. 공공데이터는 갱신 주기가 제각각이라, 언제 받은 것인지 모르면 나중에 비교할 수 없습니다.

CSV를 utf-8-sig로 저장하는 이유는 엑셀에서 열 때 한글이 깨지지 않게 하기 위해서입니다. 엑셀은 BOM이 없으면 시스템 기본 인코딩으로 읽으려 합니다.

키를 모든 항목에서 모아 정렬한 것도 의도가 있습니다. 레코드마다 있는 항목과 없는 항목이 다를 수 있어서, 첫 번째 레코드의 키만 쓰면 뒤쪽 데이터의 컬럼이 잘립니다.

여기서 자주 막힙니다

인증키를 코드에 그대로 적어두는 것은 나중에 문제가 됩니다. 코드를 공유하거나 저장소에 올릴 때 키가 함께 나갑니다. 환경변수나 별도 설정 파일로 분리하고, 그 파일은 공유 대상에서 제외하는 습관을 들이는 편이 좋습니다.

중간에 실패했을 때 처음부터 다시 받는 구조도 곤란합니다. 수천 페이지를 받는 중에 한 번 끊기면 그때까지의 작업이 사라집니다. 페이지 단위로 저장하거나, 실패한 페이지 번호를 따로 모아 재시도하는 쪽이 안전합니다.

한도 초과는 개발 단계에서 자주 겪습니다. 코드를 고치면서 반복 실행하다 보면 하루치를 금방 씁니다. 처음에는 numOfRows를 작게 두고 한두 페이지로만 시험한 뒤, 구조가 확정되면 전체를 돌리는 순서를 권합니다.

다음 단계

여기까지 하면 원본 데이터가 손에 들어옵니다. 그다음은 좌표계 변환, 인코딩 정리, 자유 텍스트 항목 파싱처럼 데이터 자체의 성격을 다루는 일이 남습니다.

그 부분은 API를 호출하는 것보다 시간이 더 걸립니다. 다만 한 번 정리해 두면 갱신본을 받을 때마다 같은 절차를 다시 쓸 수 있습니다.

---

자료 출처 · 공공데이터포털 오픈 API 이용 안내 및 각 API 기술문서
파라미터 이름과 응답 구조는 API마다 다르므로 해당 기술문서를 함께 확인하시기 바랍니다.