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

1from __future__ import annotations 

2 

3import functools 

4import re 

5import sys 

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

7from contextlib import AbstractAsyncContextManager, asynccontextmanager 

8from dataclasses import dataclass 

9from ipaddress import AddressValueError, IPv6Address 

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

11 

12import anyio.abc 

13 

14from starlette.types import Scope 

15 

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

17 from inspect import iscoroutinefunction 

18 from typing import TypeIs 

19else: # pragma: no cover 

20 from asyncio import iscoroutinefunction 

21 

22 from typing_extensions import TypeIs 

23 

24 

25T = TypeVar("T") 

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

27 

28# Reject characters that could make a Host header change the URL path or authority. 

29_HOST_RE = re.compile( 

30 r"^(?P<host>[a-z0-9._~%!$&'()*+,;=-]+|\[(?:(?P<ipv6>[a-f0-9]*:[a-f0-9.:]+)|" 

31 r"(?-i:v)[a-f0-9]+\.[a-z0-9._~!$&'()*+,;=:-]+)\])(?::(?P<port>[0-9]+))?$", 

32 re.IGNORECASE, 

33) 

34 

35 

36@dataclass(frozen=True, slots=True) 

37class ParsedHost: 

38 host: str 

39 port: str | None 

40 

41 @property 

42 def authority(self) -> str: 

43 return self.host if self.port is None else f"{self.host}:{self.port}" 

44 

45 @property 

46 def is_valid_port(self) -> bool: 

47 if self.port is None: 

48 return True 

49 port = self.port.lstrip("0") 

50 return len(port) <= 5 and int(port or "0") <= 65535 

51 

52 

53@overload 

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

55 

56 

57@overload 

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

59 

60 

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

62 while isinstance(obj, functools.partial): 

63 obj = obj.func 

64 

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

66 

67 

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

69 

70 

71class AwaitableOrContextManager( 

72 Awaitable[T_co], AbstractAsyncContextManager[T_co], Protocol[T_co] 

73): ... # pragma: no branch 

74 

75 

76class SupportsAsyncClose(Protocol): 

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

78 

79 

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

81 

82 

83class AwaitableOrContextManagerWrapper(Generic[SupportsAsyncCloseType]): 

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

85 

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

87 self.aw = aw 

88 

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

90 return self.aw.__await__() 

91 

92 async def __aenter__(self) -> SupportsAsyncCloseType: 

93 self.entered = await self.aw 

94 return self.entered 

95 

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

97 await self.entered.close() 

98 return None 

99 

100 

101@asynccontextmanager 

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

103 try: 

104 async with anyio.create_task_group() as tg: 

105 yield tg 

106 except BaseExceptionGroup as excs: 

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

108 raise 

109 

110 exc = excs.exceptions[0] 

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

112 raise exc from exc.__cause__ or context 

113 

114 

115def parse_host_header(host_header: str | None) -> ParsedHost | None: 

116 """Parse `host_header` into its host and port components. 

117 

118 The host preserves brackets around IP literals. Invalid headers produce `None`. 

119 """ 

120 if host_header is None: 

121 return None 

122 

123 match = _HOST_RE.fullmatch(host_header) 

124 if match is None: 

125 return None 

126 

127 ipv6 = match["ipv6"] 

128 if ipv6 is not None: 

129 try: 

130 IPv6Address(ipv6) 

131 except AddressValueError: 

132 return None 

133 

134 return ParsedHost(match["host"], match["port"]) 

135 

136 

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

138 path: str = scope["path"] 

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

140 if not root_path: 

141 return path 

142 

143 if not path.startswith(root_path): 

144 return path 

145 

146 if path == root_path: 

147 return "" 

148 

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

150 return path[len(root_path) :] 

151 

152 return path