Builder.shin

Python asyncio로 API 호출 병렬 처리 개선하기

문제 상황

매일 새벽 실행되는 데이터 동기화 배치 작업이 있었다. 외부 API를 호출해 약 2000개 항목의 정보를 가져오는데, 순차 처리로 인해 35분 정도 소요됐다.

# 기존 코드
for item_id in item_ids:
    response = requests.get(f"{API_URL}/{item_id}")
    save_to_db(response.json())

asyncio 도입

aiohttp와 asyncio를 사용해 비동기 처리로 전환했다. 다만 외부 API 부하를 고려해 동시 요청 수를 제한해야 했다.

import asyncio
import aiohttp
from asyncio import Semaphore

async def fetch_item(session, item_id, semaphore):
    async with semaphore:
        try:
            async with session.get(f"{API_URL}/{item_id}") as response:
                data = await response.json()
                await save_to_db_async(data)
        except Exception as e:
            logger.error(f"Failed to fetch {item_id}: {e}")

async def main():
    semaphore = Semaphore(20)  # 동시 요청 20개로 제한
    async with aiohttp.ClientSession() as session:
        tasks = [
            fetch_item(session, item_id, semaphore)
            for item_id in item_ids
        ]
        await asyncio.gather(*tasks, return_exceptions=True)

결과

실행 시간이 35분에서 4분으로 단축됐다. Semaphore로 동시 요청 수를 조절하니 API 측에서도 문제없이 처리됐다.

주의사항

  • return_exceptions=True를 설정해 일부 실패가 전체를 중단시키지 않도록 했다
  • DB 저장도 비동기로 처리해야 병목이 없다 (asyncpg 사용)
  • 세션 재사용으로 커넥션 오버헤드를 줄였다

동기 코드를 비동기로 바꾸는 건 단순 작업이 아니지만, I/O 바운드 작업에선 확실한 효과가 있었다.

Python asyncio로 API 호출 병렬 처리 개선하기