import functools
import sys
import typing
-from contextlib import asynccontextmanager, contextmanager
+from contextlib import asynccontextmanager
import anyio.abc
return None
-@contextmanager
-def _collapse_excgroups() -> typing.Generator[None, None, None]:
+@asynccontextmanager
+async def create_collapsing_task_group() -> typing.AsyncGenerator[anyio.abc.TaskGroup, None]:
try:
- yield
+ async with anyio.create_task_group() as tg:
+ yield tg
except BaseExceptionGroup as excs:
if len(excs.exceptions) != 1:
raise
del exc, cause, tb, context
-@asynccontextmanager
-async def create_collapsing_task_group() -> typing.AsyncGenerator[anyio.abc.TaskGroup, None]:
- with _collapse_excgroups():
- async with anyio.create_task_group() as tg:
- yield tg
-
-
def get_route_path(scope: Scope) -> str:
path: str = scope["path"]
root_path = scope.get("root_path", "")