38 lines
1.4 KiB
Python
38 lines
1.4 KiB
Python
import sys
|
|
from contextlib import contextmanager
|
|
from typing import Generator
|
|
|
|
# Divergence #3 (see sse_starlette/sse.py module docstring): vendored copy of
|
|
# starlette._utils.collapse_excgroups. We do not import the upstream version
|
|
# because it lives in a private (underscore-prefixed) Starlette module, and
|
|
# the pinned floor (starlette>=0.41.3) makes that import surface unstable.
|
|
#
|
|
# Behaviour: AnyIO v4 wraps task-group failures in an ExceptionGroup; this
|
|
# helper unwraps single-exception groups so user middleware sees the bare
|
|
# exception (matching pre-v4 anyio and Starlette's StreamingResponse).
|
|
# Solution per https://anyio.readthedocs.io/en/stable/migration.html
|
|
|
|
has_exceptiongroups = True
|
|
if sys.version_info < (3, 11):
|
|
try:
|
|
from exceptiongroup import BaseExceptionGroup # noqa: F401
|
|
except ImportError:
|
|
has_exceptiongroups = False
|
|
|
|
|
|
@contextmanager
|
|
def collapse_excgroups() -> Generator[None, None, None]:
|
|
try:
|
|
yield
|
|
except BaseException as exc:
|
|
if has_exceptiongroups:
|
|
# `ty` does not narrow BaseExceptionGroup.exceptions; the runtime
|
|
# contract is identical to starlette._utils.collapse_excgroups.
|
|
while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1: # ty: ignore[unresolved-attribute]
|
|
exc = exc.exceptions[0] # ty: ignore[unresolved-attribute]
|
|
|
|
raise exc
|
|
|
|
|
|
__all__ = ["collapse_excgroups"]
|