Builder.shin

Python asyncio로 API 호출 속도 개선하기

문제 상황

회사가 재택근무로 전환되면서 VPN 대역폭 이슈로 기존에 잘 돌아가던 배치 작업이 타임아웃을 내기 시작했다. 약 200개의 외부 API를 순차적으로 호출하는 스크립트였는데, 평소 5분 걸리던 작업이 20분 넘게 걸렸다.

기존 코드는 requests 라이브러리로 동기 방식 호출이었다.

import requests

def fetch_data(urls):
    results = []
    for url in urls:
        response = requests.get(url, timeout=10)
        results.append(response.json())
    return results

asyncio + aiohttp 적용

I/O 바운드 작업이 명확했기 때문에 asyncio로 전환했다. aiohttp를 사용해 비동기 HTTP 요청을 구현했다.

import asyncio
import aiohttp

async def fetch_one(session, url):
    async with session.get(url, timeout=10) as response:
        return await response.json()

async def fetch_data(urls):
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_one(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 실행
results = asyncio.run(fetch_data(urls))

동시 요청 수를 제한하고 싶다면 asyncio.Semaphore를 추가했다.

async def fetch_with_limit(urls, limit=20):
    semaphore = asyncio.Semaphore(limit)
    
    async def bounded_fetch(session, url):
        async with semaphore:
            return await fetch_one(session, url)
    
    async with aiohttp.ClientSession() as session:
        tasks = [bounded_fetch(session, url) for url in urls]
        return await asyncio.gather(*tasks, return_exceptions=True)

결과

  • 실행 시간: 20분 → 4분
  • 에러 핸들링이 필요해서 return_exceptions=True 옵션 활용
  • 세마포어로 동시 요청 20개로 제한해 API 서버 부하 조절

VPN 대역폭이 제한적인 상황에서도 병렬 처리로 충분히 개선할 수 있었다. 기존 스크립트 구조를 크게 바꾸지 않아도 되어 배포도 빠르게 진행했다.

Python asyncio로 API 호출 속도 개선하기