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

1from __future__ import annotations 

2 

3import asyncio 

4import functools 

5import re 

6import sys 

7import typing 

8from contextlib import contextmanager 

9 

10from starlette.types import Scope 

11 

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

13 from typing import TypeGuard 

14else: # pragma: no cover 

15 from typing_extensions import TypeGuard 

16 

17has_exceptiongroups = True 

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

19 try: 

20 from exceptiongroup import BaseExceptionGroup 

21 except ImportError: 

22 has_exceptiongroups = False 

23 

24T = typing.TypeVar("T") 

25AwaitableCallable = typing.Callable[..., typing.Awaitable[T]] 

26 

27 

28@typing.overload 

29def is_async_callable(obj: AwaitableCallable[T]) -> TypeGuard[AwaitableCallable[T]]: 

30 ... 

31 

32 

33@typing.overload 

34def is_async_callable(obj: typing.Any) -> TypeGuard[AwaitableCallable[typing.Any]]: 

35 ... 

36 

37 

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

39 while isinstance(obj, functools.partial): 

40 obj = obj.func 

41 

42 return asyncio.iscoroutinefunction(obj) or ( 

43 callable(obj) and asyncio.iscoroutinefunction(obj.__call__) 

44 ) 

45 

46 

47T_co = typing.TypeVar("T_co", covariant=True) 

48 

49 

50class AwaitableOrContextManager( 

51 typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co] 

52): 

53 ... 

54 

55 

56class SupportsAsyncClose(typing.Protocol): 

57 async def close(self) -> None: 

58 ... # pragma: no cover 

59 

60 

61SupportsAsyncCloseType = typing.TypeVar( 

62 "SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False 

63) 

64 

65 

66class AwaitableOrContextManagerWrapper(typing.Generic[SupportsAsyncCloseType]): 

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

68 

69 def __init__(self, aw: typing.Awaitable[SupportsAsyncCloseType]) -> None: 

70 self.aw = aw 

71 

72 def __await__(self) -> typing.Generator[typing.Any, None, SupportsAsyncCloseType]: 

73 return self.aw.__await__() 

74 

75 async def __aenter__(self) -> SupportsAsyncCloseType: 

76 self.entered = await self.aw 

77 return self.entered 

78 

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

80 await self.entered.close() 

81 return None 

82 

83 

84@contextmanager 

85def collapse_excgroups() -> typing.Generator[None, None, None]: 

86 try: 

87 yield 

88 except BaseException as exc: 

89 if has_exceptiongroups: 

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

91 exc = exc.exceptions[0] # pragma: no cover 

92 

93 raise exc 

94 

95 

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

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

98 route_path = re.sub(r"^" + root_path, "", scope["path"]) 

99 return route_path