Python asyncio로 API 병렬 호출 최적화하기

문제 상황

재택근무 환경에서 매일 새벽 실행되는 데이터 수집 배치 작업이 있었다. 약 500개의 외부 API를 호출해 결과를 DB에 저장하는 단순한 작업이었는데, requests 라이브러리로 순차 호출하다보니 30분 이상 소요됐다.

import requests

for item_id in items:
    response = requests.get(f'https://api.example.com/data/{item_id}')
    save_to_db(response.json())

API 응답 시간이 평균 2~3초인데 순차 실행이라 비효율적이었다.

asyncio 도입

aiohttp로 비동기 HTTP 요청을 처리하도록 변경했다.

import asyncio
import aiohttp

async def fetch_data(session, item_id):
    async with session.get(f'https://api.example.com/data/{item_id}') as response:
        data = await response.json()
        await save_to_db_async(data)

async def main():
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_data(session, item_id) for item_id in items]
        await asyncio.gather(*tasks)

asyncio.run(main())

동시 요청 수 제어

처음엔 500개를 한번에 실행했다가 API 서버에서 429 에러가 발생했다. Semaphore로 동시 요청 수를 20개로 제한했다.

semaphore = asyncio.Semaphore(20)

async def fetch_data(session, item_id):
    async with semaphore:
        async with session.get(f'https://api.example.com/data/{item_id}') as response:
            return await response.json()

결과

30분 → 5분으로 6배 단축됐다. 에러 핸들링과 재시도 로직도 추가했지만 충분히 빨라졌다. Python의 비동기 처리가 I/O 바운드 작업에선 확실히 효과적이었다.