Builder.shin

Python 비동기 처리: asyncio와 aiohttp로 API 호출 성능 개선

문제 상황

데이터 수집 스크립트에서 외부 API를 100회 이상 호출하는 작업이 있었다. requests 라이브러리를 사용한 순차 처리 방식이라 전체 실행 시간이 15초가 넘었다.

import requests

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

asyncio + aiohttp 적용

Python 3.7부터 asyncio가 안정화되었고, aiohttp가 비동기 HTTP 클라이언트로 자리잡았다. 기존 코드를 async/await 패턴으로 변경했다.

import asyncio
import aiohttp

async def fetch_one(session, url):
    async with session.get(url) 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)

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

결과

  • 100개 요청 기준 15초 → 2초로 단축
  • 동시 연결 수 제한이 필요한 경우 asyncio.Semaphore 사용
  • 에러 핸들링은 각 task별로 try-except 추가
async def fetch_one(session, url):
    try:
        async with session.get(url, timeout=5) as response:
            return await response.json()
    except asyncio.TimeoutError:
        return None
    except Exception as e:
        print(f"Error fetching {url}: {e}")
        return None

주의사항

  • CPU bound 작업에는 효과 없음 (multiprocessing 고려)
  • 외부 API의 rate limit 확인 필요
  • aiohttp는 requests와 API가 다르므로 migration 시 주의

비동기 처리가 필요한 I/O 작업에서는 asyncio가 확실한 선택지였다.

Python 비동기 처리: asyncio와 aiohttp로 API 호출 성능 개선