add helper for unwrapping exception groups with single exception

src/trio/__init__.py

@@ -28,6 +28,7 @@
     MemoryReceiveChannel as MemoryReceiveChannel,
     MemorySendChannel as MemorySendChannel,
     open_memory_channel as open_memory_channel,
+    raise_single_exception_from_group as raise_single_exception_from_group,
 )
 from ._core import (
     BrokenResourceError as BrokenResourceError,

src/trio/_channel.py

@@ -5,6 +5,7 @@
 from typing import (
     TYPE_CHECKING,
     Generic,
+    NoReturn,
 )
 
 import attrs
@@ -21,6 +22,11 @@
 
     from typing_extensions import Self
 
+import sys
+
+if sys.version_info < (3, 11):
+    from exceptiongroup import BaseExceptionGroup
+
 
 def _open_memory_channel(
     max_buffer_size: int | float,  # noqa: PYI041
@@ -440,3 +446,63 @@ async def aclose(self) -> None:
         See `MemoryReceiveChannel.close`."""
         self.close()
         await trio.lowlevel.checkpoint()
+
+
+def _raise(exc: BaseException) -> NoReturn:
+    """This helper allows re-raising an exception without __context__ being set."""
+    # cause does not need special handling, we simply avoid using `raise .. from ..`
+    __tracebackhide__ = True
+    context = exc.__context__
+    try:
+        raise exc
+    finally:
+        exc.__context__ = context
+        del exc, context
+
+
+# idk where to define this. It's a util, but exported, so _util doesn`t fit.
+# it'll be used by bg_with_channel, but is not directly related to channels.
+def raise_single_exception_from_group(
+    eg: BaseExceptionGroup[BaseException],
+) -> NoReturn:
+    """This function takes an exception group that is assumed to have at most
+    one non-cancelled exception, which it reraises as a standalone exception.
+
+    If a :exc:`KeyboardInterrupt` is encountered, a new KeyboardInterrupt is immediately
+    raised with the entire group as cause.
+
+    If the group only contains :exc:`Cancelled` it reraises the first one encountered.
+
+    It will retain context and cause of the contained exception, and entirely discard
+    the cause/context of the group(s).
+
+    If multiple non-cancelled exceptions are encountered, it raises
+    :exc:`AssertionError`.
+    """
+    cancelled_exceptions = []
+    noncancelled_exceptions = []
+
+    # subgroup/split retains excgroup structure, so we need to manually traverse
+    def _parse_excg(e: BaseException) -> None:
+        if isinstance(e, KeyboardInterrupt):
+            # immediately bail out
+            raise KeyboardInterrupt from eg
+
+        if isinstance(e, trio.Cancelled):
+            cancelled_exceptions.append(e)
+        elif isinstance(e, BaseExceptionGroup):
+            for sub_e in e.exceptions:
+                _parse_excg(sub_e)
+        else:
+            noncancelled_exceptions.append(e)
+
+    _parse_excg(eg)
+
+    if len(noncancelled_exceptions) > 1:
+        raise AssertionError(
+            "Attempted to unwrap exceptiongroup with multiple non-cancelled exceptions. This is often caused by a bug in the caller."
+        ) from eg
+    if len(noncancelled_exceptions) == 1:
+        _raise(noncancelled_exceptions[0])
+    assert cancelled_exceptions, "internal error"
+    _raise(cancelled_exceptions[0])

src/trio/_tests/test_channel.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import sys
 from typing import Union
 
 import pytest
@@ -9,6 +10,9 @@
 
 from ..testing import assert_checkpoints, wait_all_tasks_blocked
 
+if sys.version_info < (3, 11):
+    from exceptiongroup import BaseExceptionGroup, ExceptionGroup
+
 
 async def test_channel() -> None:
     with pytest.raises(TypeError):
@@ -411,3 +415,68 @@ async def do_send(s: trio.MemorySendChannel[int], v: int) -> None:
             assert await r.receive() == 1
     with pytest.raises(trio.WouldBlock):
         r.receive_nowait()
+
+
+async def test_raise_single_exception_from_group() -> None:
+    excinfo: pytest.ExceptionInfo[BaseException]
+
+    exc = ValueError("foo")
+    cause = SyntaxError("cause")
+    context = TypeError("context")
+    exc.__cause__ = cause
+    exc.__context__ = context
+    cancelled = trio.Cancelled._create()
+
+    with pytest.raises(ValueError, match="foo") as excinfo:
+        trio.raise_single_exception_from_group(ExceptionGroup("", [exc]))
+    assert excinfo.value.__cause__ == cause
+    assert excinfo.value.__context__ == context
+
+    with pytest.raises(ValueError, match="foo") as excinfo:
+        trio.raise_single_exception_from_group(
+            ExceptionGroup("", [ExceptionGroup("", [exc])])
+        )
+    assert excinfo.value.__cause__ == cause
+    assert excinfo.value.__context__ == context
+
+    with pytest.raises(ValueError, match="foo") as excinfo:
+        trio.raise_single_exception_from_group(
+            BaseExceptionGroup(
+                "", [cancelled, BaseExceptionGroup("", [cancelled, exc])]
+            )
+        )
+    assert excinfo.value.__cause__ == cause
+    assert excinfo.value.__context__ == context
+
+    # multiple non-cancelled
+    eg = ExceptionGroup("", [ValueError("foo"), ValueError("bar")])
+    with pytest.raises(
+        AssertionError,
+        match=r"^Attempted to unwrap exceptiongroup with multiple non-cancelled exceptions. This is often caused by a bug in the caller.$",
+    ) as excinfo:
+        trio.raise_single_exception_from_group(eg)
+    assert excinfo.value.__cause__ is eg
+    assert excinfo.value.__context__ is None
+
+    # keyboardinterrupt overrides everything
+    eg_ki = BaseExceptionGroup(
+        "",
+        [
+            ValueError("foo"),
+            ValueError("bar"),
+            KeyboardInterrupt("this exc doesn't get reraised"),
+        ],
+    )
+    with pytest.raises(KeyboardInterrupt, match=r"^$") as excinfo:
+        trio.raise_single_exception_from_group(eg_ki)
+    assert excinfo.value.__cause__ is eg_ki
+    assert excinfo.value.__context__ is None
+
+    # if we only got cancelled, first one is reraised
+    with pytest.raises(trio.Cancelled, match=r"^Cancelled$") as excinfo:
+        trio.raise_single_exception_from_group(
+            BaseExceptionGroup("", [cancelled, trio.Cancelled._create()])
+        )
+    assert excinfo.value is cancelled
+    assert excinfo.value.__cause__ is None
+    assert excinfo.value.__context__ is None

tox.ini

@@ -17,7 +17,7 @@ labels =
 # * move to pyproject.toml?
 #   * this means conditional deps need to be replaced
 
-# protip: install uv-tox for faster venv generation
+# protip: install tox-uv for faster venv generation
 
 [testenv]
 description = "Base environment for running tests depending on python version."