Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/starlette/_utils.py: 49%

1from __future__ import annotations 

2 

3import functools 

4import sys 

5from collections.abc import AsyncGenerator, Awaitable, Callable, Generator 

6from contextlib import AbstractAsyncContextManager, asynccontextmanager 

7from typing import Any, Generic, Protocol, TypeVar, overload 

8 

9import anyio.abc 

10 

11from starlette.types import Scope 

12 

13if sys.version_info >= (3, 13): # pragma: no cover 

14 from inspect import iscoroutinefunction 

15 from typing import TypeIs 

16else: # pragma: no cover 

17 from asyncio import iscoroutinefunction 

18 

19 from typing_extensions import TypeIs 

20 

21if sys.version_info < (3, 11): # pragma: no cover 

22 try: 

23 from exceptiongroup import BaseExceptionGroup 

24 except ImportError: 

25 

26 class BaseExceptionGroup(BaseException): # type: ignore[no-redef] 

27 pass 

28 

29 

30T = TypeVar("T") 

31AwaitableCallable = Callable[..., Awaitable[T]] 

32 

33 

34@overload 

35def is_async_callable(obj: AwaitableCallable[T]) -> TypeIs[AwaitableCallable[T]]: ... 

36 

37 

38@overload 

39def is_async_callable(obj: Any) -> TypeIs[AwaitableCallable[Any]]: ... 

40 

41 

42def is_async_callable(obj: Any) -> Any: 

43 while isinstance(obj, functools.partial): 

44 obj = obj.func 

45 

46 return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__)) 

47 

48 

49T_co = TypeVar("T_co", covariant=True) 

50 

51 

52class AwaitableOrContextManager( 

53 Awaitable[T_co], AbstractAsyncContextManager[T_co], Protocol[T_co] 

54): ... # pragma: no branch 

55 

56 

57class SupportsAsyncClose(Protocol): 

58 async def close(self) -> None: ... # pragma: no cover 

59 

60 

61SupportsAsyncCloseType = TypeVar("SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False) 

62 

63 

64class AwaitableOrContextManagerWrapper(Generic[SupportsAsyncCloseType]): 

65 __slots__ = ("aw", "entered") 

66 

67 def __init__(self, aw: Awaitable[SupportsAsyncCloseType]) -> None: 

68 self.aw = aw 

69 

70 def __await__(self) -> Generator[Any, None, SupportsAsyncCloseType]: 

71 return self.aw.__await__() 

72 

73 async def __aenter__(self) -> SupportsAsyncCloseType: 

74 self.entered = await self.aw 

75 return self.entered 

76 

77 async def __aexit__(self, *args: Any) -> None | bool: 

78 await self.entered.close() 

79 return None 

80 

81 

82@asynccontextmanager 

83async def create_collapsing_task_group() -> AsyncGenerator[anyio.abc.TaskGroup, None]: 

84 try: 

85 async with anyio.create_task_group() as tg: 

86 yield tg 

87 except BaseExceptionGroup as excs: 

88 if len(excs.exceptions) != 1: 

89 raise 

90 

91 exc = excs.exceptions[0] 

92 context = None if exc.__suppress_context__ else exc.__context__ 

93 raise exc from exc.__cause__ or context 

94 

95 

96def get_route_path(scope: Scope) -> str: 

97 path: str = scope["path"] 

98 root_path = scope.get("root_path", "") 

99 if not root_path: 

100 return path 

101 

102 if not path.startswith(root_path): 

103 return path 

104 

105 if path == root_path: 

106 return "" 

107 

108 if path[len(root_path)] == "/": 

109 return path[len(root_path) :] 

110 

111 return path