Builder.shin

Python asyncio로 API 요청 병렬 처리 성능 개선

문제 상황

재택근무 전환 후 매일 아침 실행하는 데이터 수집 스크립트가 문제였다. 외부 API 3000개를 순차 호출하면서 평균 40분이 소요됐다.

# 기존 코드
for item_id in item_ids:
    response = requests.get(f'https://api.example.com/items/{item_id}')
    save_to_db(response.json())

asyncio + aiohttp 적용

Python 3.7부터 안정화된 asyncio를 활용해 비동기 처리로 전환했다.

import asyncio
import aiohttp

async def fetch_item(session, item_id):
    try:
        async with session.get(f'https://api.example.com/items/{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():
    connector = aiohttp.TCPConnector(limit=50)  # 동시 연결 제한
    timeout = aiohttp.ClientTimeout(total=30)
    
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        tasks = [fetch_item(session, item_id) for item_id in item_ids]
        await asyncio.gather(*tasks, return_exceptions=True)

asyncio.run(main())

핵심 포인트

  1. TCPConnector limit: 서버 부하 방지를 위해 동시 연결 50개로 제한
  2. timeout 설정: 응답 없는 요청이 전체를 블로킹하지 않도록
  3. return_exceptions=True: 일부 실패가 전체를 중단시키지 않도록

결과

  • 처리 시간: 40분 → 4분
  • DB 연결 풀 조정 필요 (동시 쓰기 증가)
  • 로깅 개선으로 실패 건 추적 용이

비동기 코드 디버깅이 까다로웠지만, 실행 시간 단축 효과가 확실했다.

Python asyncio로 API 요청 병렬 처리 성능 개선