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

1from __future__ import annotations 

2 

3import json 

4import sys 

5from collections.abc import AsyncGenerator, Iterator, Mapping 

6from contextlib import aclosing 

7from http import cookies as http_cookies 

8from typing import TYPE_CHECKING, Any, Generic, NoReturn, cast 

9 

10import anyio 

11 

12from starlette._utils import AwaitableOrContextManager, AwaitableOrContextManagerWrapper 

13from starlette.datastructures import URL, Address, FormData, Headers, QueryParams, State 

14from starlette.exceptions import HTTPException 

15from starlette.formparsers import FormParser, MultiPartException, MultiPartParser 

16from starlette.types import Message, Receive, Scope, Send 

17 

18if TYPE_CHECKING: 

19 from python_multipart.multipart import parse_options_header 

20 

21 from starlette.applications import Starlette 

22 from starlette.middleware.sessions import Session 

23 from starlette.routing import Router 

24else: 

25 try: 

26 try: 

27 from python_multipart.multipart import parse_options_header 

28 except ModuleNotFoundError: # pragma: no cover 

29 from multipart.multipart import parse_options_header 

30 except ModuleNotFoundError: # pragma: no cover 

31 parse_options_header = None 

32 

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

34 from typing import TypeVar 

35else: # pragma: no cover 

36 from typing_extensions import TypeVar 

37 

38SERVER_PUSH_HEADERS_TO_COPY = { 

39 "accept", 

40 "accept-encoding", 

41 "accept-language", 

42 "cache-control", 

43 "user-agent", 

44} 

45 

46 

47def cookie_parser(cookie_string: str) -> dict[str, str]: 

48 """ 

49 This function parses a ``Cookie`` HTTP header into a dict of key/value pairs. 

50 

51 It attempts to mimic browser cookie parsing behavior: browsers and web servers 

52 frequently disregard the spec (RFC 6265) when setting and reading cookies, 

53 so we attempt to suit the common scenarios here. 

54 

55 This function has been adapted from Django 3.1.0. 

56 Note: we are explicitly _NOT_ using `SimpleCookie.load` because it is based 

57 on an outdated spec and will fail on lots of input we want to support 

58 """ 

59 cookie_dict: dict[str, str] = {} 

60 for chunk in cookie_string.split(";"): 

61 if "=" in chunk: 

62 key, val = chunk.split("=", 1) 

63 else: 

64 # Assume an empty name per 

65 # https://bugzilla.mozilla.org/show_bug.cgi?id=169091 

66 key, val = "", chunk 

67 key, val = key.strip(), val.strip() 

68 if key or val: 

69 # unquote using Python's algorithm. 

70 cookie_dict[key] = http_cookies._unquote(val) 

71 return cookie_dict 

72 

73 

74class ClientDisconnect(Exception): 

75 pass 

76 

77 

78StateT = TypeVar("StateT", bound=Mapping[str, Any] | State, default=State) 

79 

80 

81class HTTPConnection(Mapping[str, Any], Generic[StateT]): 

82 """ 

83 A base class for incoming HTTP connections, that is used to provide 

84 any functionality that is common to both `Request` and `WebSocket`. 

85 """ 

86 

87 def __init__(self, scope: Scope, receive: Receive | None = None) -> None: 

88 assert scope["type"] in ("http", "websocket") 

89 self.scope = scope 

90 

91 def __getitem__(self, key: str) -> Any: 

92 return self.scope[key] 

93 

94 def __iter__(self) -> Iterator[str]: 

95 return iter(self.scope) 

96 

97 def __len__(self) -> int: 

98 return len(self.scope) 

99 

100 # Don't use the `abc.Mapping.__eq__` implementation. 

101 # Connection instances should never be considered equal 

102 # unless `self is other`. 

103 __eq__ = object.__eq__ 

104 __hash__ = object.__hash__ 

105 

106 @property 

107 def app(self) -> Any: 

108 return self.scope["app"] 

109 

110 @property 

111 def url(self) -> URL: 

112 if not hasattr(self, "_url"): # pragma: no branch 

113 self._url = URL(scope=self.scope) 

114 return self._url 

115 

116 @property 

117 def base_url(self) -> URL: 

118 if not hasattr(self, "_base_url"): 

119 base_url_scope = dict(self.scope) 

120 # This is used by request.url_for, it might be used inside a Mount which 

121 # would have its own child scope with its own root_path, but the base URL 

122 # for url_for should still be the top level app root path. 

123 app_root_path = base_url_scope.get("app_root_path", base_url_scope.get("root_path", "")) 

124 path = app_root_path 

125 if not path.endswith("/"): 

126 path += "/" 

127 base_url_scope["path"] = path 

128 base_url_scope["query_string"] = b"" 

129 base_url_scope["root_path"] = app_root_path 

130 self._base_url = URL(scope=base_url_scope) 

131 return self._base_url 

132 

133 @property 

134 def headers(self) -> Headers: 

135 if not hasattr(self, "_headers"): 

136 self._headers = Headers(scope=self.scope) 

137 return self._headers 

138 

139 @property 

140 def query_params(self) -> QueryParams: 

141 if not hasattr(self, "_query_params"): # pragma: no branch 

142 self._query_params = QueryParams(self.scope["query_string"]) 

143 return self._query_params 

144 

145 @property 

146 def path_params(self) -> dict[str, Any]: 

147 path_params: dict[str, Any] = self.scope.get("path_params", {}) 

148 return path_params 

149 

150 @property 

151 def cookies(self) -> dict[str, str]: 

152 if not hasattr(self, "_cookies"): 

153 cookies: dict[str, str] = {} 

154 cookie_headers = self.headers.getlist("cookie") 

155 

156 for header in cookie_headers: 

157 cookies.update(cookie_parser(header)) 

158 

159 self._cookies = cookies 

160 return self._cookies 

161 

162 @property 

163 def client(self) -> Address | None: 

164 # client is a 2 item tuple of (host, port), None if missing 

165 host_port = self.scope.get("client") 

166 if host_port is not None: 

167 return Address(*host_port) 

168 return None 

169 

170 @property 

171 def session(self) -> dict[str, Any]: 

172 assert "session" in self.scope, "SessionMiddleware must be installed to access request.session" 

173 session: Session = self.scope["session"] 

174 # We keep the hasattr in case people actually use their own `SessionMiddleware` implementation. 

175 if hasattr(session, "mark_accessed"): # pragma: no branch 

176 session.mark_accessed() 

177 return session 

178 

179 @property 

180 def auth(self) -> Any: 

181 assert "auth" in self.scope, "AuthenticationMiddleware must be installed to access request.auth" 

182 return self.scope["auth"] 

183 

184 @property 

185 def user(self) -> Any: 

186 assert "user" in self.scope, "AuthenticationMiddleware must be installed to access request.user" 

187 return self.scope["user"] 

188 

189 @property 

190 def state(self) -> StateT: 

191 if not hasattr(self, "_state"): 

192 # Ensure 'state' has an empty dict if it's not already populated. 

193 self.scope.setdefault("state", {}) 

194 # Create a state instance with a reference to the dict in which it should 

195 # store info 

196 self._state = State(self.scope["state"]) 

197 return cast(StateT, self._state) 

198 

199 def url_for(self, name: str, /, **path_params: Any) -> URL: 

200 url_path_provider: Router | Starlette | None = self.scope.get("router") or self.scope.get("app") 

201 if url_path_provider is None: 

202 raise RuntimeError("The `url_for` method can only be used inside a Starlette application or with a router.") 

203 url_path = url_path_provider.url_path_for(name, **path_params) 

204 return url_path.make_absolute_url(base_url=self.base_url) 

205 

206 

207async def empty_receive() -> NoReturn: 

208 raise RuntimeError("Receive channel has not been made available") 

209 

210 

211async def empty_send(message: Message) -> NoReturn: 

212 raise RuntimeError("Send channel has not been made available") 

213 

214 

215class Request(HTTPConnection[StateT]): 

216 _form: FormData | None 

217 

218 def __init__(self, scope: Scope, receive: Receive = empty_receive, send: Send = empty_send): 

219 super().__init__(scope) 

220 assert scope["type"] == "http" 

221 self._receive = receive 

222 self._send = send 

223 self._stream_consumed = False 

224 self._is_disconnected = False 

225 self._form = None 

226 

227 @property 

228 def method(self) -> str: 

229 return cast(str, self.scope["method"]) 

230 

231 @property 

232 def receive(self) -> Receive: 

233 return self._receive 

234 

235 async def stream(self) -> AsyncGenerator[bytes, None]: 

236 if hasattr(self, "_body"): 

237 yield self._body 

238 yield b"" 

239 return 

240 if self._stream_consumed: 

241 raise RuntimeError("Stream consumed") 

242 while not self._stream_consumed: 

243 message = await self._receive() 

244 if message["type"] == "http.request": 

245 body = message.get("body", b"") 

246 if not message.get("more_body", False): 

247 self._stream_consumed = True 

248 if body: 

249 yield body 

250 elif message["type"] == "http.disconnect": # pragma: no branch 

251 self._is_disconnected = True 

252 raise ClientDisconnect() 

253 yield b"" 

254 

255 async def body(self) -> bytes: 

256 if not hasattr(self, "_body"): 

257 chunks: list[bytes] = [] 

258 async with aclosing(self.stream()) as stream: 

259 async for chunk in stream: 

260 chunks.append(chunk) 

261 self._body = b"".join(chunks) 

262 return self._body 

263 

264 async def json(self) -> Any: 

265 if not hasattr(self, "_json"): # pragma: no branch 

266 body = await self.body() 

267 self._json = json.loads(body) 

268 return self._json 

269 

270 async def _get_form( 

271 self, 

272 *, 

273 max_files: int | float = 1000, 

274 max_fields: int | float = 1000, 

275 max_part_size: int = 1024 * 1024, 

276 ) -> FormData: 

277 if self._form is None: # pragma: no branch 

278 assert parse_options_header is not None, ( 

279 "The `python-multipart` library must be installed to use form parsing." 

280 ) 

281 content_type_header = self.headers.get("Content-Type") 

282 content_type: bytes 

283 content_type, _ = parse_options_header(content_type_header) 

284 if content_type == b"multipart/form-data": 

285 try: 

286 async with aclosing(self.stream()) as stream: 

287 multipart_parser = MultiPartParser( 

288 self.headers, 

289 stream, 

290 max_files=max_files, 

291 max_fields=max_fields, 

292 max_part_size=max_part_size, 

293 ) 

294 self._form = await multipart_parser.parse() 

295 except MultiPartException as exc: 

296 if "app" in self.scope: 

297 raise HTTPException(status_code=400, detail=exc.message) 

298 raise exc 

299 elif content_type == b"application/x-www-form-urlencoded": 

300 try: 

301 async with aclosing(self.stream()) as stream: 

302 form_parser = FormParser( 

303 self.headers, 

304 stream, 

305 max_fields=max_fields, 

306 max_part_size=max_part_size, 

307 ) 

308 self._form = await form_parser.parse() 

309 except MultiPartException as exc: 

310 if "app" in self.scope: 

311 raise HTTPException(status_code=400, detail=exc.message) 

312 raise exc 

313 else: 

314 self._form = FormData() 

315 return self._form 

316 

317 def form( 

318 self, 

319 *, 

320 max_files: int | float = 1000, 

321 max_fields: int | float = 1000, 

322 max_part_size: int = 1024 * 1024, 

323 ) -> AwaitableOrContextManager[FormData]: 

324 return AwaitableOrContextManagerWrapper( 

325 self._get_form(max_files=max_files, max_fields=max_fields, max_part_size=max_part_size) 

326 ) 

327 

328 async def close(self) -> None: 

329 if self._form is not None: # pragma: no branch 

330 await self._form.close() 

331 

332 async def is_disconnected(self) -> bool: 

333 if not self._is_disconnected: 

334 message: Message = {} 

335 

336 # If message isn't immediately available, move on 

337 with anyio.CancelScope() as cs: 

338 cs.cancel() 

339 message = await self._receive() 

340 

341 if message.get("type") == "http.disconnect": 

342 self._is_disconnected = True 

343 

344 return self._is_disconnected 

345 

346 async def send_push_promise(self, path: str) -> None: 

347 if "http.response.push" in self.scope.get("extensions", {}): 

348 raw_headers: list[tuple[bytes, bytes]] = [] 

349 for name in SERVER_PUSH_HEADERS_TO_COPY: 

350 for value in self.headers.getlist(name): 

351 raw_headers.append((name.encode("latin-1"), value.encode("latin-1"))) 

352 await self._send({"type": "http.response.push", "path": path, "headers": raw_headers}) 

353 

354 async def send_early_hints(self, link: str, /, *additional_links: str) -> None: 

355 if "http.response.early_hint" in self.scope.get("extensions", {}): 

356 raw_links = [value.encode("latin-1") for value in (link, *additional_links)] 

357 await self._send({"type": "http.response.early_hint", "links": raw_links})