Draft some eg collapsing helpers

Inside a new `.trionics._beg` and exposed from the subpkg ns in
anticipation of the `strict_exception_groups=False` being removed by
`trio` in py 3.15.

Notes,
- mk an embedded single-exc "extractor" using a `BaseExceptionGroup.exceptions` length
  check, when 1 return the lone child.
- use the above in a new `@acm`, async bc it's most likely to be composed in an
  `async with` tuple-style sequence block, called `collapse_eg()` which
  acts a one line "absorber" for when the above mentioned flag is no
  logner supported by `trio.open_nursery()`.

All untested atm fwiw.. but soon to be used in our test suite(s) likely!
Tyler Goodlet 2025-03-03 18:01:16 -05:00
parent 802774641f
commit 539ae7d26d
2 changed files with 62 additions and 0 deletions

View File

@ -29,3 +29,6 @@ from ._broadcast import (
BroadcastReceiver as BroadcastReceiver,
Lagged as Lagged,
)
from ._beg import (
collapse_eg as collapse_eg,
)

View File

@ -0,0 +1,59 @@
# tractor: structured concurrent "actors".
# Copyright 2018-eternity Tyler Goodlet.
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
'''
`BaseExceptionGroup` related utils and helpers pertaining to
first-class-`trio` from a historical perspective B)
'''
from contextlib import (
# bontextmanager as cm,
asynccontextmanager as acm,
)
def maybe_collapse_eg(
beg: BaseExceptionGroup,
) -> BaseException:
'''
If the input beg can collapse to a single non-eg sub-exception,
return it instead.
'''
if len(excs := beg.exceptions) == 1:
return excs[0]
return beg
@acm
async def collapse_eg():
'''
If `BaseExceptionGroup` raised in the body scope is
"collapse-able" (in the same way that
`trio.open_nursery(strict_exception_groups=False)` works) then
only raise the lone emedded non-eg in in place.
'''
try:
yield
except* BaseException as beg:
if (
exc := maybe_collapse_eg(beg)
) is not beg:
raise exc
raise beg