Ptyhon 3.11 - New Feature : ExceptionGroup & TaskGroup
본 글은 Python 3.11에 새로 소개된 ExceptionGroup & TaskGroup에 대해 소개합니다.
Exception, asyncio에 사전 지식이 있는 독자를 대상으로 합니다.
ExceptionGroup
Python 3.11에서는 여러 예외들을 하나의 그룹으로 묶어서 동시에 처리할 수 있게 해주는 'ExceptionGroup'이라는 새로운 기능을 도입하였습니다. 이를 어떻게, 언제 사용할 수 있는지 살펴봅시다.
Python은 원칙적으로 한 번에 하나의 예외만을 처리할 수 있습니다. 그러나 발생한 여러 오류를 한꺼번에 처리하는 것이 더 유리한 경우도 있습니다. 멀티 프로세싱이나 asyncio 그룹 작업이 대표적인 예입니다. PEP-654에서는 이와 관련하여 더 다양한 사례를 찾아볼 수 있습니다.
예외 그룹은 Exception을 상속받는 클래스이기 때문에, 아래와 같이 사용할 수 있습니다.
try:
raise ExceptionGroup("group", [ValueError(654)])
except ExceptionGroup:
print("Handling ExceptionGroup")여러 에러를 명확히 구분하여 처리하는 것이 좋습니다. 아래 예제처럼 여러 에러를 ExceptionGroup으로 묶어 명확하게 처리할 수 있습니다.
try:
raise ExceptionGroup(
"group", [TypeError("str"), ValueError(654), TypeError("int")]
)
except* ValueError as eg:
print(f"Handling ValueError: {eg.exceptions}")
except* TypeError as eg:
print(f"Handling TypeError: {eg.exceptions}")다만, except* 구문을 사용하여 모든 오류를 처리해야 합니다. 아래 예제에서는 ValueError만 처리하고 나머지 TypeError는 처리되지 않아 예외가 발생합니다.
try:
raise ExceptionGroup(
"group", [TypeError("str"), ValueError(654), TypeError("int")]
# 개인적으로 위 코드가 너무 인위적이라서 실제로 이런 에러가 발생할 예시가 와닿지 않았는데
# 다음 문단에서 예시가 나옵니다!
)
except* ValueError as eg:
print(f"Handling ValueError: {eg.exceptions}") | ExceptionGroup: group (2 sub-exceptions)
+-+---------------- 1 ----------------
| TypeError: str
+---------------- 2 ----------------
| TypeError: int
+------------------------------------ExceptionGroup은 기존의 예외 처리를 대체하는 것이 아니라, 여러 예외를 동시에 처리할 필요가 있을 때 유용합니다.
실제로 Python 사용자가 ExceptionGroup을 사용할 일은 많지 않을 수 있습니다. 그러나 Python 3.11이 널리 사용됨에 따라, 의존하는 패키지에서 예외 그룹을 발생시킬 수 있으므로, 애플리케이션에서 이를 처리할 필요가 있을 수 있기에 알아두면 좋을 듯 합니다 :)
TaskGroup
TaskGroup은비동기 작업을 동시에 실행하고 관리하는 데 사용되는 새로운 기능입니다. 물론 이전에도 asyncio.gather이라는 기능이 있었죠. asyncio.gather과의 주요 차이점을 살펴봅시다.
-
에러 핸들링 :
TaskGroup은 작업 중 하나라도 실패하면 즉시 다른 모든 작업을 취소합니다. 이는asyncio.gather와 비교하여 에러를 더 일찍 포착하고, 불필요한 작업 실행을 방지할 수 있게 해줍니다.gather를 사용할 때는return_exceptions파라미터를True로 설정하지 않는 이상, 모든 작업이 완료된 후에야 예외를 던집니다. 또한 TaskGroup은 Exeption Group으로 에러를 관리합니다. -
자동 취소 :
TaskGroup내에서 실행되는 모든 작업은TaskGroup이 종료될 때 자동으로 취소됩니다. 이는 추가적인 취소 로직을 작성할 필요없이, 코드를 더 간결하고 안전하게 만듭니다. -
동적 작업 추가 :
TaskGroup을 사용하면 그룹 실행 중에도 새로운 작업을 동적으로 추가할 수 있습니다. 이는asyncio.gather에서는 불가능한데,gather는 호출 시점에 모든 작업을 알고 있어야 합니다. -
손쉬운 관리 : TaskGroup을 사용하면 with 문 내에서 발생하는 모든 작업의 예외를 처리하고 적절히 집계하고, 작업 그룹의 생명 주기를 명확하게 관리할 수 있습니다.
gather은 비동기 작업을 개별 작업을 묶어서 처리하고 각각 영향을 안 받지만 TaskGroup은 여러 작업이 묶여서 하나의 작업이 되는 느낌입니다. (연대 책임) 여기서 그럼 ExceptionGroup의 힘이 발휘되지 않을까요?
우선 asyncio.gather를 사용하는 코드를 살펴봅시다.
import asyncio
import sys
import colorama
from colorama import Cursor
colorama.init()
async def print_at(row, text):
print(Cursor.POS(1, 1 + row) + str(text))
await asyncio.sleep(0.03)
async def count_lines_in_file(file_num, file_name):
counter_text = f"{file_name[:20]:<20} "
with open(file_name, mode="rt", encoding="utf-8") as file:
for line_num, _ in enumerate(file, start=1):
counter_text += "□"
await print_at(file_num, counter_text)
await print_at(file_num, f"{counter_text} ({line_num})")
async def count_all_files(file_names):
tasks = [
asyncio.create_task(count_lines_in_file(file_num, file_name))
for file_num, file_name in enumerate(file_names, start=1)
]
await asyncio.gather(*tasks)
if __name__ == "__main__":
asyncio.run(count_all_files(sys.argv[1:]))
아래와 같은 명령어를 실행하면 not_utf8.txt, empty_file.txt 이 두 파일에러 에러가 발생하리라는 것을 예상 할 수 있습니다.
python count_taskgroup.py not_utf8.txt empty_file.txt하지만 실제 에러는 1개만 출력됩니다.
UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe5 in position 2: invalid continuation byte이제 TaskGroup으로 바꿔 본 다음 동일한 명령어를 입력해 봅시다.
async def count_all_files(file_names):
async with asyncio.TaskGroup() as tg:
for file_num, file_name in enumerate(file_names, start=1):
tg.create_task(count_lines_in_file(file_num, file_name))
---
python count_taskgroup.py not_utf8.txt empty_file.txt
+ Exception Group Traceback (most recent call last):
| ...
| ExceptionGroup: unhandled errors in a TaskGroup (2 sub-exceptions)
+-+---------------- 1 ----------------
| Traceback (most recent call last):
| File "count_taskgroup.py", line 18, in count_lines_in_file
| for line_num, _ in enumerate(file, start=1):
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
| UnicodeDecodeError: 'utf-8' codec can't decode byte 0xe5 in position 2:
| invalid continuation byte
+---------------- 2 ----------------
| Traceback (most recent call last):
| File "count_taskgroup.py", line 21, in count_lines_in_file
| await print_at(file_num, f"{counter_text} ({line_num})")
| ^^^^^^^^
| UnboundLocalError: cannot access local variable 'line_num' where it is
| not associated with a value
+------------------------------------두 개의 에러가 모두 출력 되고 있습니다.
조금 더 쉬운 예제를 살펴봅시다.
from asyncio import TaskGroup
import asyncio
async def task(n):
if n % 2 == 0:
await asyncio.sleep(0.1)
raise ValueError(f"Value error in task {n}")
return f"Task {n} completed successfully"
async def main():
try:
async with TaskGroup() as tg:
for i in range(4):
tg.create_task(task(i))
except* ValueError as e:
print(e.exceptions)
# (ValueError('Value error in task 0'), ValueError('Value error in task 2'))
asyncio.run(main())async def main():
tasks = [task(i) for i in range(4)]
result = await asyncio.gather(*tasks, return_exceptions=True)
print(result)
# [ValueError('Value error in task 0'), 'Task 1 completed successfully', ValueError('Value error in task 2'), 'Task 3 completed successfully']
asyncio.run(main())결론
확실히 asyncio.gather는 독립적인 코루틴을 단순히 모아둔다는 느낌을 줍니다. 반면, TaskGroup은 여러 작업을 묶어 하나의 큰 작업 단위로 만듭니다. 그룹이 하나의 작업 단위가 되었을 때, TaskGroup을 사용하면 에러 핸들링 로직이나 생명 주기를 보다 편리하게 관리할 수 있겠습니다.
다음 글은 Python의 Asyncio를 주제로 해보고자 합니다. 긴 글 읽어주셔서 감사합니다 :)
