طبقه‌بندی شکست TaskGroup در پایتون؛ ExceptionGroup و حفظ شاخه‌های خطا

وقتی یک دستهٔ اندازه‌گیری چند مشکل مستقل دارد، چاپ نام یک استثنا تصویر کاملی نمی‌دهد. باید بدانیم خطا زیر کدام دسته و رک قرار گرفته، کدام بخش سیاست رسیدگی مشخص دارد و کدام خطا باید به فراخواننده برسد. در این درس، درخت ExceptionGroup را در یک اجرای واقعی TaskGroup می‌سازیم و ساختارش را هنگام طبقه‌بندی حفظ می‌کنیم. داده‌ها و خرابی‌ها همگی محلی و ساختگی‌اند.

۱. درختِ خطاهای یک دستهٔ اندازه‌گیری

فرض کنید گزارش کنترل ولتاژ، دو رک غربی و شرقی دارد. رک غربی کانال A با مقدار ۷٫۲۵ ولت و کانال B با نشانگر خرابی خواندن دارد. رک شرقی شامل C با منفی شش ولت و D با نشانگر اشکال رمزگشاست. محدودهٔ پذیرفته‌شدهٔ آموزشی منفی پنج تا مثبت پنج ولت، با دو سرِ شامل، تعریف شده است. مقادیر A و C بیرون آن‌اند.

برای دادهٔ نامعتبر کلاس SensorDataFault، برای خرابی خواندن کلاس InjectedReadFault و برای اشکال ناشناختهٔ رمزگشا RuntimeError می‌سازیم. نام دسته، واحد V و نام رک‌ها در گره‌های گروه قرار می‌گیرند؛ شناسهٔ کانال در پیام برگ است. چهار شکست داریم، اما فقط دو رک. تعداد فرزندان مستقیمِ یک گروه، لزوماً تعداد همهٔ خطاهای زیر آن نیست.

بخش ۸٫۹ آموزش رسمی پایتون ExceptionGroup را برای گزارش چند استثنای مستقل معرفی می‌کند. در مدل ما، استقلال یعنی بررسی یک رکورد مانع بررسی رکورد دیگر نیست. خرابی‌ها الزاماً هم‌زمان رخ نداده‌اند؛ هر بررسیِ محلی تمام می‌شود و خطایش برای انتشارِ گروهی نگه داشته می‌شود. این رفتار برای یک دستهٔ محدود، سیاست صریح نویسنده است.

عمداً فقط یک worker در TaskGroup داریم. این worker همهٔ چهار بررسی محدود را بدون await انجام می‌دهد و سپس گروهِ تودرتو را بالا می‌برد. بنابراین ساخته‌شدن چهار برگ به مسابقهٔ زمان‌بندی چند فرزند وابسته نیست. این انتخاب، نمایش عملیِ ظرف خطا و طبقه‌بندی آن است؛ ادعای شکست هم‌زمان چند وظیفه، اجرای موازی یا جمع‌آوری تمام شکست‌های احتمالیِ یک سامانه را نداریم.

۲. جداسازیِ برگ‌ها با حفظِ شاخه‌ها

طبق مرجع ExceptionGroup، متد subgroup بخش مطابق شرط و split دو بخش مطابق و باقی‌مانده را می‌دهد. شاخه‌های لازم و فرادادهٔ گروه حفظ می‌شوند و شاخهٔ خالی کنار می‌رود. به همین دلیل خروجیِ خرابی خواندن هنوز از پوشش TaskGroup، دستهٔ ولتاژ و رک غربی عبور می‌کند؛ فقط برگ B در آن باقی است.

تابع classify ابتدا خرابی خواندن را جدا می‌کند؛ سپس از باقی‌مانده، خطاهای داده را برمی‌دارد. هرچه مانده، unknown_faults است. کلاس‌های دو دستهٔ شناخته‌شده عمداً هم‌پوشانی ندارند. اگر در طراحی دیگری کلاس‌ها اشتراک داشته باشند، ترتیب جداسازی مهم می‌شود؛ نمی‌توان یک برگ را هم‌زمان در دو بخشِ همین افراز مطالبه کرد.

اگر هیچ برگی برای یک دسته نباشد، مقدار None داریم؛ گروه تهی نمی‌سازیم. نتیجهٔ تابع یک dataclass منجمد با سه ارجاع است، اما استثناهای زیر آن کپی عمیق نشده‌اند. انجماد ظرف به معنای تغییرناپذیری همهٔ اشیای داخل نیست. کد، برگ‌ها و اطلاعات اصلی را برای ادامهٔ بررسی نگه می‌دارد؛ متن JSON فقط نمای قابل‌چاپِ جداگانه است.

این‌جا شرط را از نوع استثنا می‌دهیم. شرط callable می‌تواند روی خودِ گروه داخلی هم صدق کند و یک زیرشاخهٔ کامل را انتخاب کند؛ پس اگر فقط برگ خاصی را می‌خواهیم، نوشتن شرطی که هر گرهٔ گروه را هم درست بداند نتیجه را عوض می‌کند. همین نکته دلیل مناسبی برای انتخاب کلاس‌های دقیق و ساده در این مثال است.

۳. کد کاملِ TaskGroup و طبقه‌بندیِ شکست

فایل را با نام exceptiongroup_failures.py ذخیره و با python exceptiongroup_failures.py اجرا کنید. اجرای واقعی Python 3.14.5 بوده است؛ منابع رسمیِ خوانده‌شده نسخهٔ 3.14.8 دارند. هیچ بستهٔ بیرونی، ورودی شبکه، sleep یا تلاش مجدد نداریم. شناسه‌ها و مقادیرِ سناریو ثابت‌اند؛ این فایل، رابط عمومیِ دریافت هر نوع رکورد حسگر نیست.

"""Deterministic offline exception-tree classification; no retries or IO."""
import asyncio
from collections import Counter
from dataclasses import dataclass
import json
import math
import platform


class SensorDataFault(ValueError):
    pass


class InjectedReadFault(TimeoutError):
    """A synthetic failure category, not an elapsed timeout."""


@dataclass(frozen=True)
class Classification:
    read_faults: ExceptionGroup | None
    data_faults: ExceptionGroup | None
    unknown_faults: ExceptionGroup | None


def classify(group):
    # Fatal errors and cancellation are deliberately outside this API.
    if not isinstance(group, ExceptionGroup):
        raise TypeError('ordinary ExceptionGroup required')
    reads, rest = group.split(InjectedReadFault)
    if rest is None:
        return Classification(reads, None, None)
    data, unknown = rest.split(SensorDataFault)
    return Classification(reads, data, unknown)


def tree(exc):
    """JSON-safe structural view; never replaces the actual exception objects."""
    if exc is None:
        return None
    if isinstance(exc, BaseExceptionGroup):
        return {'group': exc.message, 'children': [tree(e) for e in exc.exceptions]}
    return {'type': type(exc).__name__, 'message': str(exc)}


def leaves(exc):
    if exc is None:
        return []
    if isinstance(exc, BaseExceptionGroup):
        return [leaf for child in exc.exceptions for leaf in leaves(child)]
    return [exc]


def path_oracle(exc):
    """Independent iterative tree walk; does not call classify/split/subgroup."""
    found = []
    pending = [((), exc)]
    while pending:
        path, current = pending.pop()
        if isinstance(current, BaseExceptionGroup):
            for child in current.exceptions:
                pending.append((path + (current.message,), child))
        else:
            found.append((path, type(current).__name__, str(current), id(current)))
    return Counter(found)


def inspect(channel, value):
    if value == 'read_unavailable':
        raise InjectedReadFault(f'{channel}: injected read unavailable')
    if value == 'decoder_bug':
        raise RuntimeError(f'{channel}: injected decoder bug')
    if type(value) not in (int, float) or not math.isfinite(value) or not -5 <= value <= 5:
        raise SensorDataFault(f'{channel}: outside finite [-5, 5] V contract')


async def inspect_batch():
    # One worker completes all independent local inspections before raising.
    # There is no await in this finite worker and no sibling scheduling race.
    racks = []
    for rack, records in (
        ('rack-west', (('A', 7.25), ('B', 'read_unavailable'))),
        ('rack-east', (('C', -6.0), ('D', 'decoder_bug'))),
    ):
        errors = []
        for channel, value in records:
            try:
                inspect(channel, value)
            except (SensorDataFault, InjectedReadFault, RuntimeError) as exc:
                errors.append(exc)  # Retained for grouped propagation.
        if errors:
            racks.append(ExceptionGroup(rack, errors))
    if racks:
        raise ExceptionGroup('batch-voltage-V', racks)


async def actual_taskgroup_failure():
    try:
        async with asyncio.TaskGroup() as group:
            group.create_task(inspect_batch())
    except ExceptionGroup as exc:
        return exc  # Observation boundary; callers must classify/propagate.
    raise AssertionError('fixture must fail')


def handle_known(group, observations):
    """Known categories have explicit reporting policy; unknown auto-propagates."""
    try:
        raise group
    except* InjectedReadFault as matching:
        observations['read'] = tree(matching)
    except* SensorDataFault as matching:
        observations['data'] = tree(matching)


async def cancellation_probe():
    entered = asyncio.Event()
    blocker = asyncio.Event()
    closed = []

    async def worker():
        try:
            entered.set()
            await blocker.wait()
        finally:
            closed.append('cleanup')

    task = asyncio.create_task(worker())
    await entered.wait()
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        # This is an isolated test boundary, not application suppression.
        return task.cancelled() and closed == ['cleanup']
    return False


async def run_async():
    group = await actual_taskgroup_failure()
    result = classify(group)
    categories = (result.read_faults, result.data_faults, result.unknown_faults)
    original_leaves = leaves(group)
    checks = {}
    checks['actual_taskgroup_wrapper_and_four_leaves'] = (
        isinstance(group, ExceptionGroup) and len(group.exceptions) == 1
        and group.exceptions[0].message == 'batch-voltage-V' and len(original_leaves) == 4)
    checks['hand_category_counts'] = [len(leaves(x)) for x in categories] == [1, 2, 1]
    checks['known_class_subgroup_agrees'] = (
        tree(group.subgroup(InjectedReadFault)) == tree(result.read_faults))
    observed_paths = sum((path_oracle(x) for x in categories if x is not None), Counter())
    checks['iterative_path_oracle_and_leaf_identity'] = observed_paths == path_oracle(group)
    checks['disjoint_categories_without_leaf_loss'] = (
        Counter(id(e) for x in categories for e in leaves(x)) == Counter(map(id, original_leaves)))
    expected_paths = {
        ('batch-voltage-V', 'rack-west', 'InjectedReadFault'),
        ('batch-voltage-V', 'rack-west', 'SensorDataFault'),
        ('batch-voltage-V', 'rack-east', 'SensorDataFault'),
        ('batch-voltage-V', 'rack-east', 'RuntimeError'),
    }
    # Omit implementation-chosen TaskGroup outer message from this hand oracle.
    checks['hand_nested_branch_paths'] = {
        path[1:] + (kind,) for path, kind, message, identity in path_oracle(group)
    } == expected_paths
    observations = {}
    propagated = None
    try:
        handle_known(group, observations)
    except ExceptionGroup as exc:
        propagated = exc
    checks['except_star_handles_known_and_propagates_unknown'] = (
        observations == {'read': tree(result.read_faults), 'data': tree(result.data_faults)}
        and tree(propagated) == tree(result.unknown_faults)
        and leaves(propagated)[0] is leaves(result.unknown_faults)[0])
    homogeneous = ExceptionGroup('only-data', [SensorDataFault('A')])
    homogeneous_result = classify(homogeneous)
    checks['homogeneous_and_absent_category_none'] = (
        homogeneous_result.read_faults is None and homogeneous_result.unknown_faults is None
        and tree(homogeneous_result.data_faults) == tree(homogeneous))
    all_unknown = ExceptionGroup('only-unknown', [KeyError('unexpected')])
    checks['all_unknown_preserved'] = tree(classify(all_unknown).unknown_faults) == tree(all_unknown)
    before = tree(group)
    group.add_note('offline inspection fixture')
    cause = ValueError('fixture source context')
    group.__cause__ = cause
    annotated = classify(group)
    checks['group_metadata_and_original_structure'] = (
        tree(group) == before
        and all(x.__traceback__ is group.__traceback__ and x.__cause__ is cause
                and x.__notes__ == group.__notes__ for x in
                (annotated.read_faults, annotated.data_faults, annotated.unknown_faults)))
    cancellation = asyncio.CancelledError('cancel fixture')
    fatal_group = BaseExceptionGroup('control', [SensorDataFault('A'), cancellation])
    rejected = []
    for invalid in (fatal_group, SensorDataFault('plain'), None):
        try:
            classify(invalid)
        except TypeError:
            rejected.append(True)
    checks['ordinary_group_boundary_rejects_control_flow'] = len(rejected) == 3
    checks['base_group_auto_narrows_when_ordinary'] = type(BaseExceptionGroup('plain', [ValueError('x')])) is ExceptionGroup
    try:
        ExceptionGroup('invalid', [cancellation])
    except TypeError:
        checks['exceptiongroup_cannot_wrap_cancelled_error'] = True
    checks['real_cancellation_observed_after_finally'] = await cancellation_probe()
    known_observations = {}
    handle_known(homogeneous, known_observations)
    checks['explicit_known_only_report_policy'] = set(known_observations) == {'data'}
    assert all(type(v) is bool and v for v in checks.values()), checks
    return {
        'python_version': platform.python_version(), 'stdlib_only': True,
        'scope': 'one actual TaskGroup child; complete finite local inspection; no IO/sleep/retry',
        'sample_unit': 'V', 'allowed_closed_interval': [-5, 5],
        'category_leaf_counts': {'read': 1, 'data': 2, 'unknown': 1},
        'original_failure_tree': before,
        'read_failure_tree': tree(result.read_faults),
        'data_failure_tree': tree(result.data_faults),
        'unknown_propagated_tree': tree(propagated),
        'retries_performed': 0, 'checks': checks,
    }


def run_example():
    return asyncio.run(run_async())


if __name__ == '__main__':
    print(json.dumps(run_example(), indent=2, ensure_ascii=False))

مرجع کامل TaskGroup انتظار پایان فرزندان و جمع‌کردن خطاهای غیرلغو را توضیح می‌دهد. در این اجرا، خطای تودرتوی یک فرزند زیر یک پوشش بیرونی ظاهر می‌شود. متن پوشش را چاپ می‌کنیم، اما oracle دستی به عبارت خاصِ انتخاب‌شدهٔ پیاده‌سازی وابسته نیست. هیچ فرض عمومی دربارهٔ ترتیب خطاهای فرزندانِ هم‌زمان از این خروجی استخراج نمی‌شود.

تابع actual_taskgroup_failure یک مرز مشاهدهٔ آموزشی است: گروه را برای آزمودن برمی‌گرداند. دریافت این شیء به معنی موفقیت دسته نیست. در مسیر handle_known، دو بند except* فقط سیاست گزارشِ دسته‌های شناخته‌شده را اجرا می‌کنند و خطای ناشناخته خودکار بیرون می‌رود. بلوک بیرونیِ آزمون آن را برای مقایسه می‌گیرد؛ برنامهٔ کاربردی نباید صرفاً به علت چاپ‌شدن گزارش، شکست باقی‌مانده را موفق اعلام کند.

InjectedReadFault از TimeoutError مشتق شده، ولی نشانگرش مستقیماً استثنا می‌سازد. هیچ مهلتی سپری نشده و هیچ عملیات خواندن واقعی رخ نداده است. این دسته حتی به معنای «تلاش مجدد امن است» نیست؛ پیش‌شرط تکرار، اثر جانبی و سیاست بازیابی در این درس تعریف نشده‌اند. شمار retries_performed دقیقاً صفر است و نام کلاس مجوز اجرای دوبارهٔ کار نمی‌دهد.

۴. مرجع مسیرها و انتشارِ خطای ناشناخته

جواب دستی سه شمار دارد: یک خرابی خواندن، دو خطای داده و یک خطای ناشناخته. انتظار داریم شاخهٔ داده هم رک غربی با A و هم رک شرقی با C را نگه دارد. شاخهٔ خواندن فقط B و باقی‌مانده فقط D را دارد. گزارش خروجی، این شاخه‌ها را کامل نشان می‌دهد؛ به فهرست تخت نام کلاس‌ها بسنده نمی‌کنیم.

{
  "python_version": "3.14.5",
  "stdlib_only": true,
  "scope": "one actual TaskGroup child; complete finite local inspection; no IO/sleep/retry",
  "sample_unit": "V",
  "allowed_closed_interval": [
    -5,
    5
  ],
  "category_leaf_counts": {
    "read": 1,
    "data": 2,
    "unknown": 1
  },
  "original_failure_tree": {
    "group": "unhandled errors in a TaskGroup",
    "children": [
      {
        "group": "batch-voltage-V",
        "children": [
          {
            "group": "rack-west",
            "children": [
              {
                "type": "SensorDataFault",
                "message": "A: outside finite [-5, 5] V contract"
              },
              {
                "type": "InjectedReadFault",
                "message": "B: injected read unavailable"
              }
            ]
          },
          {
            "group": "rack-east",
            "children": [
              {
                "type": "SensorDataFault",
                "message": "C: outside finite [-5, 5] V contract"
              },
              {
                "type": "RuntimeError",
                "message": "D: injected decoder bug"
              }
            ]
          }
        ]
      }
    ]
  },
  "read_failure_tree": {
    "group": "unhandled errors in a TaskGroup",
    "children": [
      {
        "group": "batch-voltage-V",
        "children": [
          {
            "group": "rack-west",
            "children": [
              {
                "type": "InjectedReadFault",
                "message": "B: injected read unavailable"
              }
            ]
          }
        ]
      }
    ]
  },
  "data_failure_tree": {
    "group": "unhandled errors in a TaskGroup",
    "children": [
      {
        "group": "batch-voltage-V",
        "children": [
          {
            "group": "rack-west",
            "children": [
              {
                "type": "SensorDataFault",
                "message": "A: outside finite [-5, 5] V contract"
              }
            ]
          },
          {
            "group": "rack-east",
            "children": [
              {
                "type": "SensorDataFault",
                "message": "C: outside finite [-5, 5] V contract"
              }
            ]
          }
        ]
      }
    ]
  },
  "unknown_propagated_tree": {
    "group": "unhandled errors in a TaskGroup",
    "children": [
      {
        "group": "batch-voltage-V",
        "children": [
          {
            "group": "rack-east",
            "children": [
              {
                "type": "RuntimeError",
                "message": "D: injected decoder bug"
              }
            ]
          }
        ]
      }
    ]
  },
  "retries_performed": 0,
  "checks": {
    "actual_taskgroup_wrapper_and_four_leaves": true,
    "hand_category_counts": true,
    "known_class_subgroup_agrees": true,
    "iterative_path_oracle_and_leaf_identity": true,
    "disjoint_categories_without_leaf_loss": true,
    "hand_nested_branch_paths": true,
    "except_star_handles_known_and_propagates_unknown": true,
    "homogeneous_and_absent_category_none": true,
    "all_unknown_preserved": true,
    "group_metadata_and_original_structure": true,
    "ordinary_group_boundary_rejects_control_flow": true,
    "base_group_auto_narrows_when_ordinary": true,
    "exceptiongroup_cannot_wrap_cancelled_error": true,
    "real_cancellation_observed_after_finally": true,
    "explicit_known_only_report_policy": true
  }
}

مرجع مستقل با پشتهٔ صریح روی درخت اصلی حرکت می‌کند و برای هر برگ، مسیر پیام‌های گروه، نوع، متن و هویت شیء را ثبت می‌کند. این مرجع split، subgroup یا classify را فراخوانی نمی‌کند. مجموع چندمجموعه‌های سه خروجی باید با مرجع اصلی برابر شود. شمارنده باعث می‌شود تکرار احتمالیِ مسیر یا پیام، هنگام مقایسه به‌اشتباه حذف نشود.

پانزده گروه بررسی اجرا شده‌اند: شمار دستی، مسیرهای رک‌ها، برابر بودن subgroup و بخش خواندن، نداشتن برگِ گمشده، دستهٔ همگن، دستهٔ تماماً ناشناخته و حفظ فراداده از جملهٔ آن‌ها هستند. در بررسی فراداده، traceback و cause با هویت مرجع و notes با مقدار مقایسه می‌شوند. این مشاهده مربوط به همین گروه‌هاست؛ از آن بازسازی خودکار گزارش کاملِ traceback از JSON نتیجه نمی‌گیریم.

مشخصات کامل PEP 654 تطبیق بازگشتیِ except* و انتشار قسمت بی‌رسیدگی را روشن می‌کند. در کد، دو بند شناخته‌شده هر کدام یک بار برای زیرگروه خود اجرا می‌شوند و RuntimeError زیر رک شرقی بیرون می‌آید. این با except معمول فرق دارد که می‌تواند کل شیء ExceptionGroup را بگیرد، بی‌آنکه برگ دلخواه را جدا کرده باشد.

برای مقدمات عمر فرزند، سقف reader و پاک‌سازی، آزمون TaskGroup و لغوِ حسگر مکمل این درس است؛ مسئلهٔ تازهٔ ما حفظ ساختار شکست هنگام افراز است. اگر نتیجه به رویداد تبدیل شود، گزارش ساختاریافتهٔ رویداد حسگر انتخاب فیلد و قالب را جدا بررسی می‌کند. نمای درخت اینجا، سیاست ثبت امنِ پیام‌های دلخواه نیست.

۵. چهار پرسش پاسخ‌دار دربارهٔ لغو و رسیدگی

پرسش اول: سه دسته یعنی سه برگ خطا؟ پاسخ: خیر؛ در این مثال چهار برگ داریم و دستهٔ داده دو برگ را نگه می‌دارد. دستهٔ غایب نیز None است. تعداد دسته، رک، فرزند مستقیم و برگ باید جدا شمرده شوند.

پرسش دوم: پس از دو بند except*، خطای D چه می‌شود؟ پاسخ: چون هیچ بندِ شناخته‌شده آن را رسیدگی نکرده، با شاخه‌های مربوط بیرون می‌رود. آزمون هویت همان برگ را هم مقایسه می‌کند؛ ناشناخته با خطای داده یا گزارش موفق جایگزین نشده است.

پرسش سوم: چرا classify هر BaseExceptionGroup را نمی‌پذیرد؟ پاسخ: ExceptionGroup فقط برگ‌های Exception را می‌پذیرد؛ CancelledError از این دامنه بیرون است. گروهِ مختلط با لغو در مرز رابط رد می‌شود. مسیر کاربرد نیز فقط ExceptionGroup را می‌گیرد؛ KeyboardInterrupt و SystemExit سیاست ویژهٔ خود TaskGroup را دارند و اینجا به خطای عادی تبدیل نمی‌شوند.

پرسش چهارم: آیا آزمایش لغو، درخواست توقف را می‌بلعد؟ پاسخ: worker در finally پاک‌سازی را ثبت می‌کند و لغو به awaitکننده می‌رسد؛ فقط مرز مستقل آزمون آن را مشاهده می‌کند. Event آغاز worker را قطعی می‌کند و زمان اندازه‌گیری نمی‌شود. منابع در ۲۶ مهر ۱۴۰۵ خوانده شدند؛ PEP در ۲۲ فوریهٔ ۲۰۲۱ ایجاد شده و مربوط به Python 3.11 است. این تاریخ با نسخهٔ مستندات و اجرای حاضر یکی نیست.

منابع و مطالعهٔ بیشتر

• Python 3.14؛ قرارداد ExceptionGroup، split و subgroup
• Python 3.14؛ بخش کامل TaskGroup
• Python 3.14؛ آموزش ۸٫۹، چند استثنای مستقل
• PEP 654؛ مشخصات کامل گروه استثنا و except*

بررسی منابع: ۲۶ مهر ۱۴۰۵. تحلیل‌ها و پیشنهادهای اجرایی، جمع‌بندی نویسنده‌اند.