Python 비동기 처리에서 asyncio.gather 대신 TaskGroup 사용하기

문제 상황

여러 외부 API를 병렬로 호출하는 배치 작업을 운영 중이었다. asyncio.gather를 사용했는데, 일부 API가 실패해도 나머지는 계속 실행되어야 했다.

results = await asyncio.gather(
    fetch_api_a(),
    fetch_api_b(),
    fetch_api_c(),
    return_exceptions=True
)

return_exceptions=True로 처리했지만, 결과 리스트에서 예외와 정상 값을 구분하는 로직이 지저분했다.

TaskGroup 도입

Python 3.11에서 추가된 asyncio.TaskGroup을 사용해 구조를 개선했다.

async def process_apis():
    async with asyncio.TaskGroup() as tg:
        task_a = tg.create_task(fetch_api_a())
        task_b = tg.create_task(fetch_api_b())
        task_c = tg.create_task(fetch_api_c())
    
    return {
        'a': task_a.result(),
        'b': task_b.result(),
        'c': task_c.result()
    }

TaskGroup은 컨텍스트 매니저로 동작하며, 내부 태스크 중 하나라도 실패하면 다른 태스크를 취소하고 예외를 전파한다.

개별 예외 처리

실패한 API만 로깅하고 나머지는 진행해야 하는 경우, 각 태스크에서 예외를 잡았다.

async def safe_fetch(name, coro):
    try:
        return await coro
    except Exception as e:
        logger.error(f"{name} failed: {e}")
        return None

async with asyncio.TaskGroup() as tg:
    tasks = [
        tg.create_task(safe_fetch('api_a', fetch_api_a())),
        tg.create_task(safe_fetch('api_b', fetch_api_b())),
    ]

결과

• 예외 처리 로직이 명확해졌다
• 각 태스크의 결과를 변수로 바로 접근 가능
• 구조화된 동시성 패턴 적용

Python 3.11 이상 환경이라면 TaskGroup 사용을 권장한다.