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

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@typing.overload 

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

34 

35 

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

37 while isinstance(obj, functools.partial): 

38 obj = obj.func 

39 

40 return asyncio.iscoroutinefunction(obj) or (callable(obj) and asyncio.iscoroutinefunction(obj.__call__)) 

41 

42 

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

44 

45 

46class AwaitableOrContextManager(typing.Awaitable[T_co], typing.AsyncContextManager[T_co], typing.Protocol[T_co]): ... 

47 

48 

49class SupportsAsyncClose(typing.Protocol): 

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

51 

52 

53SupportsAsyncCloseType = typing.TypeVar("SupportsAsyncCloseType", bound=SupportsAsyncClose, covariant=False) 

54 

55 

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

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

58 

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

60 self.aw = aw 

61 

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

63 return self.aw.__await__() 

64 

65 async def __aenter__(self) -> SupportsAsyncCloseType: 

66 self.entered = await self.aw 

67 return self.entered 

68 

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

70 await self.entered.close() 

71 return None 

72 

73 

74@contextmanager 

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

76 try: 

77 yield 

78 except BaseException as exc: 

79 if has_exceptiongroups: 

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

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

82 

83 raise exc 

84 

85 

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

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

88 route_path = re.sub(r"^" + root_path + r"(?=/|$)", "", scope["path"]) 

89 return route_path