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

1from __future__ import annotations 

2 

3import functools 

4import sys 

5from collections.abc import Awaitable, Callable, Generator 

6from contextlib import AbstractAsyncContextManager, contextmanager 

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

8 

9from starlette.types import Scope 

10 

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

12 from inspect import iscoroutinefunction 

13 from typing import TypeIs 

14else: # pragma: no cover 

15 from asyncio import iscoroutinefunction 

16 

17 from typing_extensions import TypeIs 

18 

19has_exceptiongroups = True 

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

21 try: 

22 from exceptiongroup import BaseExceptionGroup # type: ignore[unused-ignore,import-not-found] 

23 except ImportError: 

24 has_exceptiongroups = False 

25 

26T = TypeVar("T") 

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

28 

29 

30@overload 

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

32 

33 

34@overload 

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

36 

37 

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

39 while isinstance(obj, functools.partial): 

40 obj = obj.func 

41 

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

43 

44 

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

46 

47 

48class AwaitableOrContextManager( 

49 Awaitable[T_co], AbstractAsyncContextManager[T_co], Protocol[T_co] 

50): ... # pragma: no branch 

51 

52 

53class SupportsAsyncClose(Protocol): 

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

55 

56 

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

58 

59 

60class AwaitableOrContextManagerWrapper(Generic[SupportsAsyncCloseType]): 

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

62 

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

64 self.aw = aw 

65 

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

67 return self.aw.__await__() 

68 

69 async def __aenter__(self) -> SupportsAsyncCloseType: 

70 self.entered = await self.aw 

71 return self.entered 

72 

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

74 await self.entered.close() 

75 return None 

76 

77 

78@contextmanager 

79def collapse_excgroups() -> Generator[None, None, None]: 

80 try: 

81 yield 

82 except BaseException as exc: 

83 if has_exceptiongroups: # pragma: no cover 

84 while isinstance(exc, BaseExceptionGroup) and len(exc.exceptions) == 1: 

85 exc = exc.exceptions[0] 

86 

87 raise exc 

88 

89 

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

91 path: str = scope["path"] 

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

93 if not root_path: 

94 return path 

95 

96 if not path.startswith(root_path): 

97 return path 

98 

99 if path == root_path: 

100 return "" 

101 

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

103 return path[len(root_path) :] 

104 

105 return path