Python asyncio로 외부 API 호출 성능 개선하기

문제 상황

매일 새벽에 돌아가는 데이터 동기화 배치 작업이 점점 느려지고 있었다. 외부 API를 호출해서 데이터를 가져오는 작업인데, 처리할 데이터가 300개를 넘어가면서 15분 이상 소요되기 시작했다.

기존 코드는 단순한 반복문으로 구성되어 있었다.

import requests

def sync_data(ids):
    results = []
    for id in ids:
        response = requests.get(f'https://api.example.com/data/{id}')
        results.append(response.json())
    return results

asyncio와 aiohttp로 전환

I/O 바운드 작업이라 동시성 처리가 효과적일 것으로 판단했다. asyncio와 aiohttp를 사용해 비동기로 전환했다.

import asyncio
import aiohttp

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

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

# 실행
results = asyncio.run(sync_data(ids))

동시 요청 수 제한

처음에는 모든 요청을 동시에 보냈는데, 외부 API 서버에서 rate limit 에러가 발생했다. Semaphore로 동시 요청 수를 제한했다.

async def sync_data(ids, max_concurrent=20):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def fetch_with_limit(session, id):
        async with semaphore:
            return await fetch_data(session, id)
    
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_with_limit(session, id) for id in ids]
        results = await asyncio.gather(*tasks)
    return results

결과

300개 데이터 처리 시간이 15분에서 2분으로 줄었다. 동시 요청 수는 20개로 설정했고, 외부 API의 응답 시간이 평균 400ms 정도인 점을 고려하면 합리적인 결과였다.

기존 레거시 코드를 건드리는 게 부담스러웠지만, 실제로는 핵심 로직만 async/await로 감싸면 되어서 리팩토링이 어렵지 않았다. 다만 라이브러리는 requests 대신 aiohttp를 사용해야 했다.