Python asyncio로 API 병렬 호출 성능 개선

문제 상황

외부 파트너사 API에서 10만 건의 데이터를 가져와 DB에 적재하는 마이그레이션 작업이 있었다. 기존 코드는 requests를 사용한 순차 처리였고, 완료까지 8시간 이상 소요됐다.

import requests

def fetch_data(item_id):
    response = requests.get(f'https://api.example.com/items/{item_id}')
    return response.json()

for item_id in item_ids:
    data = fetch_data(item_id)
    save_to_db(data)

asyncio 도입

Python 3.7부터 asyncio가 안정화되면서 비동기 HTTP 호출을 시도했다. aiohttp를 사용해 동시에 50개씩 요청을 보내도록 수정했다.

import asyncio
import aiohttp

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

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

# 50개씩 배치 처리
for i in range(0, len(item_ids), 50):
    batch = item_ids[i:i+50]
    results = asyncio.run(process_batch(batch))
    save_batch_to_db(results)

결과

• 작업 시간: 8시간 → 1.5시간
• API 서버 부하 고려해 동시 요청 수를 50개로 제한
• Connection pool 재사용으로 TCP handshake 오버헤드 감소

처음엔 asyncio 러닝커브가 부담스러웠지만, I/O bound 작업에서는 확실히 효과적이었다. 다만 디버깅이 동기 코드보다 까다로워 적절한 로깅이 필수였다.