[Python] ExceptionGroup & TaskGroup

Created

April 29, 2025

Created by

DaEun Kim

Tags

Python

Property

레퍼런스

파이썬 3.10 버전까지는 아래와 같이 예외 A 가 발생함으로써 예외 B가 발생하는 경우 예외 A는 traceback 으로 접근하지 않으면 디버깅하기 어렵다.

게다가 traceback 모듈에서 지원하는 print_exception() 같은 함수로 숨겨진 예외를 출력하려면 함수가 요구하는 파라미터들을 매번 넘겨줘야 한다.

import traceback


class A(Exception):
    pass


class B(Exception):
    pass



def raise_a():
    raise A("exception A raised")



def raise_b():
    try:
        raise_a()
    except A:
        raise B("exception B raised")

try:
    raise_b()
except B as e:
    print(f"Type of exception:: {type(e)}")
    print("::::::Traceback:::::")
    traceback.print_exception(type(e), e, e.__traceback__)

$ python tb.py
Type of exception:: <class '__main__.B'>
::::::Traceback:::::
Traceback (most recent call last):
  File "/Users/daeun/bookk-bookk/apps/tb.py", line 19, in raise_b
    raise_a()
  File "/Users/daeun/bookk-bookk/apps/tb.py", line 13, in raise_a
    raise A("exception A raised")
A: exception A raised

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/Users/daeun/bookk-bookk/apps/tb.py", line 24, in <module>
    raise_b()
  File "/Users/daeun/bookk-bookk/apps/tb.py", line 21, in raise_b
    raise B("exception B raised")
B: exception B raised

이런 번거로움을 개선하고자 3.11 버전부터 ExceptionGroup 이 도입되었다.

Exception 클래스를 상속받은 ExceptionGroup 은 서로 관련없는 여러 예외들과 서브그룹을 묶어준다.

import traceback

eg = ExceptionGroup(
     "one",
     [
         TypeError(1),
         ExceptionGroup(
             "two",
              [TypeError(2), ValueError(3)]
         ),
         ExceptionGroup(
              "three",
               [OSError(4)]
         )
     ]
 )

print("======== print all exceptions groups ========")
traceback.print_exception(eg)

is_type_error = lambda e: isinstance(e, TypeError)
type_errors = eg.subgroup(is_type_error)  # eg 내에 TypeError 객체가 있는 서브그룹만 걸러냄

print("======== print TypeError exception groups ========")
traceback.print_exception(type_errors)

print("======== print exceptions in TypeError exception groups ========")
print(type_errors.exceptions) # 그룹 내에 있는 예외 및 서브그룹 출력

print("======== print exceptions in non-TypeError exception groups ========")
type_errors, other_errors = eg.split(TypeError)  # TypeError 객체가 있는 서브그룹과 아닌 서브그룹 분리
traceback.print_exception(other_errors)

$ python exception_groups.py
======== print all exceptions groups ========
  | ExceptionGroup: one (3 sub-exceptions)
  +-+---------------- 1 ----------------
    | TypeError: 1
    +---------------- 2 ----------------
    | ExceptionGroup: two (2 sub-exceptions)
    +-+---------------- 1 ----------------
      | TypeError: 2
      +---------------- 2 ----------------
      | ValueError: 3
      +------------------------------------
    +---------------- 3 ----------------
    | ExceptionGroup: three (1 sub-exception)
    +-+---------------- 1 ----------------
      | OSError: 4
      +------------------------------------
======== print TypeError exception groups ========
  | ExceptionGroup: one (2 sub-exceptions)
  +-+---------------- 1 ----------------
    | TypeError: 1
    +---------------- 2 ----------------
    | ExceptionGroup: two (1 sub-exception)
    +-+---------------- 1 ----------------
      | TypeError: 2
      +------------------------------------
======== print exceptions in TypeError exception groups ========
(TypeError(1), ExceptionGroup('two', [TypeError(2)]))
======== print exceptions in non-TypeError exception groups ========
  | ExceptionGroup: one (2 sub-exceptions)
  +-+---------------- 1 ----------------
    | ExceptionGroup: two (1 sub-exception)
    +-+---------------- 1 ----------------
      | ValueError: 3
      +------------------------------------
    +---------------- 2 ----------------
    | ExceptionGroup: three (1 sub-exception)
    +-+---------------- 1 ----------------
      | OSError: 4
      +------------------------------------

ExceptionGroup 을 활용하면 숨겨진 여러 예외들을 잡을 수 있다. 그룹 내에 있는 예외들을 잡으려면 except* 구문을 사용한다.

class A(Exception):
    pass


class B(Exception):
    pass


class C(Exception):
    pass



def raise_a():
    raise ExceptionGroup("exception groups 1", [A("exception A raised")])



def raise_b():
    try:
        raise_a()
    except ExceptionGroup as eg:
        raise ExceptionGroup("exception groups 2", [eg, B("exception B raised")])


def raise_c():
    try:
        raise_b()
    except ExceptionGroup as eg:
        raise ExceptionGroup("exception groups 3", [eg, C("exception C raised")])


try:
    raise_c()
except* A as a:
    print(a)
    print("A raised")
except* B as b:
    print(b)
    print("B raised")
except* C as c:
    print(c)
    print("C raised")

$ python tb_exception_groups.py
exception groups 3 (1 sub-exception)
A raised
exception groups 3 (1 sub-exception)
B raised
exception groups 3 (1 sub-exception)
C raised

ExceptionGroup 의 또 다른 활용 사례로 TaskGroup 이 있다. TaskGroup 은 gather() 처럼 다수의 비동기 태스크를 수행한다.

import asyncio
import random


class A(Exception):
    pass


class B(Exception):
    pass


async def raise_error_or_not(raise_type_error: bool = False, raise_value_error: bool = False):
    await asyncio.sleep(random.uniform(0.1, 0.5))
    print("raise_error_or_not called")
    if raise_type_error:
        raise TypeError("Type error raised")
    if raise_value_error:
        raise ValueError("Value error raised")


async def run():
    try:
        async with asyncio.TaskGroup() as tg:
            tg.create_task(raise_error_or_not())
            tg.create_task(raise_error_or_not())
            tg.create_task(raise_error_or_not(raise_type_error=True))
            tg.create_task(raise_error_or_not(raise_type_error=True))
            tg.create_task(raise_error_or_not(raise_value_error=True))
    except ExceptionGroup as e:
        print(e.exceptions)

asyncio.run(run())

$ python task_group.py  # 1차 실행
raise_error_or_not called
(ValueError('Value error raised'),)

$ python task_group.py  # 2차 실행
raise_error_or_not called
raise_error_or_not called
(TypeError('Type error raised'), TypeError('Type error raised'))

랜덤하게 시간차를 두고 예외를 발생시키는 함수를 TaskGroup 의 태스크로 등록하면 위와 같은 결과가 나온다. async context 에서 빠져나온 뒤 (TaskGroup.__aexit__() 호출 된 뒤) TaskGroup 에 등록된 태스크들을 수행하는 과정에서 일부 태스크에 예외가 발생하면 나머지 태스크들은 수행되지 않으므로 5개보다 적은 갯수의 태스크가 실행된다.

이 때 발생한 예외들은 ExceptionGroup 으로 묶여서 raise 된다.

TaskGroup 은 먼저 수행되는 태스크에 예외가 발생하면 나머지 태스크들이 취소되지만 gather() 의 경우 gather(..., raise_exceptions=True) 옵션을 통해 일부 태스크에 예외가 발생하더라도 다른 태스크를 수행할 수 있다.

gather(..., raise_exceptions=False) 옵션의 경우 일부 태스크에 예외가 발생하면 다른 태스크들이 수행되지 않는다는 점에서 TaskGroup 과 동일하지만, 다수의 예외 처리는 TaskGroup 에 비해 간결하지 못하다.

import asyncio


class A(Exception):
    pass


class B(Exception):
    pass


async def raise_a():
    await asyncio.sleep(random.uniform(0.1, 0.3)
    print("A called")
    raise ExceptionGroup("A", [A("Exception A raised")])


async def raise_b():
    await asyncio.sleep(random.uniform(0.1, 0.3)
    print("B called")
    raise ExceptionGroup("B", [B("Exception B raised")])


async def return_c(c: int):
    print("C called")
    return c


async def run():
    tasks = [
        asyncio.create_task(raise_a()),
        asyncio.create_task(raise_b()),
        asyncio.create_task(return_c(3))
    ]

    try:
        await asyncio.gather(*tasks, return_exceptions=False)
    except* A as eg:
        for exc in eg.exceptions:
            print(f"exception::{type(exc).__name__}, msg::{exc}")
    except* B as eg:
        for exc in eg.exceptions:
            print(f"exception::{type(exc).__name__}, msg::{exc}")


asyncio.run(run())

$ python gather_exception_groups.py
C called
A called
exception::A, msg::Exception A raised

서로의 호출 결과에 의존하지 않는 여러 태스크들을 수행할 때는 gather(..., raise_exceptions=True) 가 적합한데, 아래와 같은 조건문을 작성해야 한다.

results = await asyncio.gather(raise_a(), raise_b(), return_c(3), return_exceptions=True)
for r in results:
  if isinstance(r, Exception):
		  print(f"exception::{type(r)}, msg::{r}")
  else:
		  print(r)