Python 비동기 작업에서 asyncio.gather의 예외 처리

문제 상황

외부 API 10개를 동시에 호출하는 데이터 수집 스크립트를 작성했다. 그런데 하나의 API라도 실패하면 전체 작업이 중단되는 문제가 발생했다.

async def fetch_all():
    tasks = [fetch_api(url) for url in urls]
    results = await asyncio.gather(*tasks)
    return results

하나의 API가 타임아웃되면 나머지 9개의 정상 결과도 받지 못하고 전체가 실패했다.

해결 방법

asyncio.gather의 return_exceptions=True 옵션을 사용했다. 이 옵션을 주면 예외가 발생해도 중단되지 않고, 예외 객체를 결과 리스트에 포함시킨다.

async def fetch_all():
    tasks = [fetch_api(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    success = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    
    print(f"성공: {len(success)}, 실패: {len(failed)}")
    return success

결과

일부 API가 실패해도 나머지 작업은 정상적으로 완료되었다. 로깅을 통해 어떤 API가 실패했는지 추적할 수 있게 되었고, 재시도 로직도 실패한 항목만 대상으로 구현할 수 있었다.

배치 작업의 안정성이 크게 개선되었다. 특히 외부 의존성이 많은 작업에서는 부분 실패를 허용하는 것이 전체 시스템의 신뢰성을 높이는 방법이었다.