파이썬 asyncio Task exception was never retrieved 원인과 예외 회수 방법

Task exception was never retrieved는 create_task()로 실행을 맡긴 Task의 예외를 호출부가 await, result() 또는 exception()으로 회수하지 않았다는 뜻입니다. 반면 coroutine was never awaited는 coroutine 객체를 만들고도 실행 예약이나 await를 하지 않았다는 뜻이므로, 먼저 두 경고를 구분해야 합니다. 자동화 코드에서는 작업의 성공 여부를 호출부가 알아야 하는지에 따라 await·TaskGroup·gather·background Task 중 방식을 선택하면 됩니다.

Table of Contents

먼저 두 경고를 구분합니다

다음처럼 coroutine을 호출한 결과만 보관하면 아직 실행이 예약되지 않습니다.

async def collect():
    return {"ok": True}

result = collect()  # coroutine 객체만 생성

이 경우에는 await collect()처럼 직접 기다리거나 asyncio.create_task(collect())처럼 Task로 예약해야 합니다. 실행되지 않은 coroutine을 그대로 두면 coroutine was never awaited 경고의 원인이 될 수 있습니다.

반대로 create_task()가 반환한 Task가 실제로 실행되었지만 내부에서 예외가 났고, 그 Task를 아무도 기다리지 않았다면 Task exception was never retrieved가 남을 수 있습니다. 즉, 첫 번째는 실행 시작 자체가 빠진 문제이고 두 번째는 실행된 작업의 실패를 호출부가 확인하지 않은 문제입니다.

완료가 필요한 작업은 await로 예외를 회수합니다

결과를 저장하거나 다음 API 호출의 입력으로 사용해야 하는 작업이라면 background로 흘려보내지 말고 완료를 기다리는 편이 안전합니다.

import asyncio

async def collect():
    return {"items": 3}

async def main():
    task = asyncio.create_task(collect())
    data = await task
    print(data)

asyncio.run(main())

await task는 작업의 결과를 받는 동시에 coroutine에서 발생한 예외를 현재 흐름으로 다시 전달합니다. 여러 작업을 함께 관리해야 한다면 작업의 생명주기를 한곳에서 보이게 만드는 구조가 중요합니다.

실패하면 함께 정리할 작업은 TaskGroup으로 묶습니다

서로 독립적으로 보이지만 모두 성공해야 후속 처리를 진행할 수 있는 작업이라면 TaskGroup을 우선 검토할 수 있습니다.

import asyncio

async def fetch_users():
    return ["kim", "lee"]

async def fetch_orders():
    return [101, 102]

async def main():
    async with asyncio.TaskGroup() as group:
        users = group.create_task(fetch_users())
        orders = group.create_task(fetch_orders())

    print(users.result(), orders.result())

asyncio.run(main())

TaskGroup 블록을 빠져나오는 시점에 포함된 작업들의 완료를 관리하므로, 작업을 만들고 잊어버리는 구조를 줄일 수 있습니다. 한 작업의 실패가 전체 작업 흐름에 어떤 영향을 주어야 하는지 먼저 정한 뒤 사용해야 하며, 실패를 무조건 숨기는 도구로 보면 안 됩니다.

결과를 모아 처리할 때는 gather의 반환 방식을 확인합니다

여러 coroutine의 결과를 한 번에 수집하려면 asyncio.gather()가 읽기 쉬울 수 있습니다.

results = await asyncio.gather(
    fetch_users(),
    fetch_orders(),
)

호출부가 결과를 받아야 하는 작업이라면 gather() 자체를 await해야 합니다. 예외를 결과 목록으로 구분해 수집해야 하는 설계라면 return_exceptions=True를 검토할 수 있지만, 그때는 반환 목록에 예외가 섞일 수 있으므로 성공 값과 실패 값을 분리해 처리해야 합니다. 실패를 정상 결과처럼 저장하지 않도록 이 판단을 코드에 드러내는 것이 좋습니다.

정말 독립적인 background Task라면 참조와 callback을 둡니다

호출부가 결과를 기다릴 필요는 없지만 작업이 끝났는지와 실패 여부는 관찰해야 한다면, Task를 집합에 보관하고 완료 callback에서 예외를 회수하는 패턴을 사용할 수 있습니다.

import asyncio

background_tasks = set()

def report_done(task: asyncio.Task) -> None:
    background_tasks.discard(task)
    try:
        task.result()
    except asyncio.CancelledError:
        pass
    except Exception as exc:
        print(f"background task failed: {exc}")

def start_background(coro):
    task = asyncio.create_task(coro)
    background_tasks.add(task)
    task.add_done_callback(report_done)
    return task

이 구조의 핵심은 두 가지입니다. Task의 참조를 보관해 생명주기를 관리하고, 완료 시 result()를 호출해 성공 결과 또는 예외를 확인하는 것입니다. 다만 background 작업의 실패가 데이터 정합성에 영향을 준다면 callback에서 로그만 남길 것이 아니라 재시도, 알림 또는 상위 흐름으로 전달하는 정책까지 정해야 합니다.

운영 코드에서 확인할 순서

경고가 coroutine was never awaited인지 Task exception was never retrieved인지 확인합니다.
create_task()로 만든 Task가 반드시 완료되어야 하는 작업인지 결정합니다.
반드시 완료되어야 하면 await, 여러 작업의 생명주기를 함께 관리해야 하면 TaskGroup, 결과를 모아야 하면 gather()를 선택합니다.
독립 background 작업이면 참조 집합과 done callback을 두고 result() 또는 exception()으로 실패를 관찰합니다.
원인을 찾기 어렵다면 asyncio debug mode를 켜서 Task 생성 위치를 확인합니다.

asyncio.run(main(), debug=True) 또는 PYTHONASYNCIODEBUG=1 같은 debug 설정은 개발 단계에서 Task가 어디서 만들어졌는지 추적하는 데 도움이 됩니다. 운영에서는 debug 설정 자체보다, 어떤 작업이 성공해야 자동화 전체를 성공으로 판단하는지와 실패를 어디에 기록하고 알릴지를 먼저 정해 두는 편이 중요합니다.

마무리

이 경고를 없애는 가장 빠른 방법은 모든 Task에 무조건 callback을 붙이는 것이 아닙니다. 작업 결과가 필요하면 await, 함께 성공해야 하는 묶음이면 TaskGroup, 여러 결과를 모으면 gather(), 정말 독립적인 작업만 참조 집합과 done callback을 사용하는 식으로 작업의 책임 범위를 코드에 맞추는 것이 핵심입니다.

다음에 확인할 질문

asyncio.create_task()와 await는 실행 시점과 예외 전달에서 어떻게 다른가?
여러 API 작업 중 하나가 실패할 때 TaskGroup과 gather() 중 어떤 동작이 필요한가?
background Task의 실패를 로그만 남기지 않고 재시도나 알림으로 연결하려면 어디에 정책을 둘 것인가?