Python 비동기 작업에서 asyncio.gather vs TaskGroup 선택 기준

배경

레거시 API 서버에서 여러 외부 서비스를 병렬로 호출하는 로직이 있었다. 기존에는 asyncio.gather를 사용했는데, Python 3.11 업그레이드를 하면서 TaskGroup으로 마이그레이션할지 검토했다.

기존 코드의 문제

async def fetch_user_data(user_id):
    results = await asyncio.gather(
        get_profile(user_id),
        get_orders(user_id),
        get_preferences(user_id),
        return_exceptions=True
    )
    # 일부 실패해도 계속 진행
    return process_results(results)

return_exceptions=True를 써서 일부 실패를 허용했지만, 어떤 태스크가 실패했는지 추적이 어려웠다. 로깅도 각 함수 내부에서 처리해야 했다.

TaskGroup 적용

async def fetch_user_data(user_id):
    results = {}
    
    async with asyncio.TaskGroup() as tg:
        tasks = {
            'profile': tg.create_task(get_profile(user_id)),
            'orders': tg.create_task(get_orders(user_id)),
            'preferences': tg.create_task(get_preferences(user_id))
        }
    
    # 모든 태스크 완료 후
    for key, task in tasks.items():
        try:
            results[key] = task.result()
        except Exception as e:
            logger.error(f"Failed to fetch {key}: {e}")
            results[key] = None
    
    return results

주요 차이점

에러 핸들링: TaskGroup은 하나라도 실패하면 모든 태스크를 취소한다. 부분 실패를 허용하려면 각 태스크를 개별 try-except로 감싸야 한다.
구조적 동시성: TaskGroup은 컨텍스트 매니저를 벗어나기 전에 모든 태스크가 완료됨을 보장한다. gather는 awaited만 되면 끝이다.
취소 동작: TaskGroup은 자동으로 남은 태스크를 취소하지만, gather는 명시적으로 처리해야 한다.

결론

우리 케이스는 부분 실패를 허용해야 해서 gather를 유지했다. 하지만 신규 배치 작업에서는 TaskGroup을 적용했다. 하나라도 실패하면 전체를 롤백해야 하는 경우 TaskGroup이 더 안전했다.

선택 기준은 명확하다. 모두 성공해야 하면 TaskGroup, 부분 성공을 허용하면 gather.