Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/web_exceptions.py: 82%

1import warnings 

2from collections.abc import Iterable 

3from http import HTTPStatus 

4from typing import Any 

5 

6from multidict import CIMultiDict 

7from yarl import URL 

8 

9from . import hdrs 

10from .helpers import CookieMixin 

11from .typedefs import LooseHeaders, StrOrURL 

12 

13__all__ = ( 

14 "HTTPException", 

15 "HTTPError", 

16 "HTTPRedirection", 

17 "HTTPSuccessful", 

18 "HTTPOk", 

19 "HTTPCreated", 

20 "HTTPAccepted", 

21 "HTTPNonAuthoritativeInformation", 

22 "HTTPNoContent", 

23 "HTTPResetContent", 

24 "HTTPPartialContent", 

25 "HTTPMove", 

26 "HTTPMultipleChoices", 

27 "HTTPMovedPermanently", 

28 "HTTPFound", 

29 "HTTPSeeOther", 

30 "HTTPNotModified", 

31 "HTTPUseProxy", 

32 "HTTPTemporaryRedirect", 

33 "HTTPPermanentRedirect", 

34 "HTTPClientError", 

35 "HTTPBadRequest", 

36 "HTTPUnauthorized", 

37 "HTTPPaymentRequired", 

38 "HTTPForbidden", 

39 "HTTPNotFound", 

40 "HTTPMethodNotAllowed", 

41 "HTTPNotAcceptable", 

42 "HTTPProxyAuthenticationRequired", 

43 "HTTPRequestTimeout", 

44 "HTTPConflict", 

45 "HTTPGone", 

46 "HTTPLengthRequired", 

47 "HTTPPreconditionFailed", 

48 "HTTPRequestEntityTooLarge", 

49 "HTTPRequestURITooLong", 

50 "HTTPUnsupportedMediaType", 

51 "HTTPRequestRangeNotSatisfiable", 

52 "HTTPExpectationFailed", 

53 "HTTPMisdirectedRequest", 

54 "HTTPUnprocessableEntity", 

55 "HTTPFailedDependency", 

56 "HTTPUpgradeRequired", 

57 "HTTPPreconditionRequired", 

58 "HTTPTooManyRequests", 

59 "HTTPRequestHeaderFieldsTooLarge", 

60 "HTTPUnavailableForLegalReasons", 

61 "HTTPServerError", 

62 "HTTPInternalServerError", 

63 "HTTPNotImplemented", 

64 "HTTPBadGateway", 

65 "HTTPServiceUnavailable", 

66 "HTTPGatewayTimeout", 

67 "HTTPVersionNotSupported", 

68 "HTTPVariantAlsoNegotiates", 

69 "HTTPInsufficientStorage", 

70 "HTTPNotExtended", 

71 "HTTPNetworkAuthenticationRequired", 

72) 

73 

74 

75class NotAppKeyWarning(UserWarning): 

76 """Warning when not using AppKey in Application.""" 

77 

78 

79############################################################ 

80# HTTP Exceptions 

81############################################################ 

82 

83 

84class HTTPException(CookieMixin, Exception): 

85 # You should set in subclasses: 

86 # status = 200 

87 

88 status_code = -1 

89 empty_body = False 

90 default_reason = "" # Initialized at the end of the module 

91 

92 def __init__( 

93 self, 

94 *, 

95 headers: LooseHeaders | None = None, 

96 reason: str | None = None, 

97 text: str | None = None, 

98 content_type: str | None = None, 

99 ) -> None: 

100 if reason is None: 

101 reason = self.default_reason 

102 elif "\r" in reason or "\n" in reason: 

103 raise ValueError("Reason cannot contain \\r or \\n") 

104 

105 if text is None: 

106 if not self.empty_body: 

107 text = f"{self.status_code}: {reason}" 

108 else: 

109 if self.empty_body: 

110 warnings.warn( 

111 f"text argument is deprecated for HTTP status {self.status_code} " 

112 "since 4.0 and scheduled for removal in 5.0 (#3462)," 

113 "the response should be provided without a body", 

114 DeprecationWarning, 

115 stacklevel=2, 

116 ) 

117 

118 if headers is not None: 

119 real_headers = CIMultiDict(headers) 

120 else: 

121 real_headers = CIMultiDict() 

122 

123 if content_type is not None: 

124 if not text: 

125 warnings.warn( 

126 "content_type without text is deprecated " 

127 "since 4.0 and scheduled for removal in 5.0 " 

128 "(#3462)", 

129 DeprecationWarning, 

130 stacklevel=2, 

131 ) 

132 real_headers[hdrs.CONTENT_TYPE] = content_type 

133 elif hdrs.CONTENT_TYPE not in real_headers and text: 

134 real_headers[hdrs.CONTENT_TYPE] = "text/plain" 

135 

136 self._reason = reason 

137 self._text = text 

138 self._headers = real_headers 

139 self.args = () 

140 

141 def __bool__(self) -> bool: 

142 return True 

143 

144 @property 

145 def status(self) -> int: 

146 return self.status_code 

147 

148 @property 

149 def reason(self) -> str: 

150 return self._reason 

151 

152 @property 

153 def text(self) -> str | None: 

154 return self._text 

155 

156 @property 

157 def headers(self) -> "CIMultiDict[str]": 

158 return self._headers 

159 

160 def __str__(self) -> str: 

161 return self.reason 

162 

163 def __repr__(self) -> str: 

164 return f"<{self.__class__.__name__}: {self.reason}>" 

165 

166 __reduce__ = object.__reduce__ 

167 

168 def __getnewargs__(self) -> tuple[Any, ...]: 

169 return self.args 

170 

171 

172class HTTPError(HTTPException): 

173 """Base class for exceptions with status codes in the 400s and 500s.""" 

174 

175 

176class HTTPRedirection(HTTPException): 

177 """Base class for exceptions with status codes in the 300s.""" 

178 

179 

180class HTTPSuccessful(HTTPException): 

181 """Base class for exceptions with status codes in the 200s.""" 

182 

183 

184class HTTPOk(HTTPSuccessful): 

185 status_code = 200 

186 

187 

188class HTTPCreated(HTTPSuccessful): 

189 status_code = 201 

190 

191 

192class HTTPAccepted(HTTPSuccessful): 

193 status_code = 202 

194 

195 

196class HTTPNonAuthoritativeInformation(HTTPSuccessful): 

197 status_code = 203 

198 

199 

200class HTTPNoContent(HTTPSuccessful): 

201 status_code = 204 

202 empty_body = True 

203 

204 

205class HTTPResetContent(HTTPSuccessful): 

206 status_code = 205 

207 empty_body = True 

208 

209 

210class HTTPPartialContent(HTTPSuccessful): 

211 status_code = 206 

212 

213 

214############################################################ 

215# 3xx redirection 

216############################################################ 

217 

218 

219class HTTPMove(HTTPRedirection): 

220 def __init__( 

221 self, 

222 location: StrOrURL, 

223 *, 

224 headers: LooseHeaders | None = None, 

225 reason: str | None = None, 

226 text: str | None = None, 

227 content_type: str | None = None, 

228 ) -> None: 

229 if not location: 

230 raise ValueError("HTTP redirects need a location to redirect to.") 

231 super().__init__( 

232 headers=headers, reason=reason, text=text, content_type=content_type 

233 ) 

234 self._location = URL(location) 

235 self.headers["Location"] = str(self.location) 

236 

237 @property 

238 def location(self) -> URL: 

239 return self._location 

240 

241 

242class HTTPMultipleChoices(HTTPMove): 

243 status_code = 300 

244 

245 

246class HTTPMovedPermanently(HTTPMove): 

247 status_code = 301 

248 

249 

250class HTTPFound(HTTPMove): 

251 status_code = 302 

252 

253 

254# This one is safe after a POST (the redirected location will be 

255# retrieved with GET): 

256class HTTPSeeOther(HTTPMove): 

257 status_code = 303 

258 

259 

260class HTTPNotModified(HTTPRedirection): 

261 # FIXME: this should include a date or etag header 

262 status_code = 304 

263 empty_body = True 

264 

265 

266class HTTPUseProxy(HTTPMove): 

267 # Not a move, but looks a little like one 

268 status_code = 305 

269 

270 

271class HTTPTemporaryRedirect(HTTPMove): 

272 status_code = 307 

273 

274 

275class HTTPPermanentRedirect(HTTPMove): 

276 status_code = 308 

277 

278 

279############################################################ 

280# 4xx client error 

281############################################################ 

282 

283 

284class HTTPClientError(HTTPError): 

285 pass 

286 

287 

288class HTTPBadRequest(HTTPClientError): 

289 status_code = 400 

290 

291 

292class HTTPUnauthorized(HTTPClientError): 

293 status_code = 401 

294 

295 

296class HTTPPaymentRequired(HTTPClientError): 

297 status_code = 402 

298 

299 

300class HTTPForbidden(HTTPClientError): 

301 status_code = 403 

302 

303 

304class HTTPNotFound(HTTPClientError): 

305 status_code = 404 

306 

307 

308class HTTPMethodNotAllowed(HTTPClientError): 

309 status_code = 405 

310 

311 def __init__( 

312 self, 

313 method: str, 

314 allowed_methods: Iterable[str], 

315 *, 

316 headers: LooseHeaders | None = None, 

317 reason: str | None = None, 

318 text: str | None = None, 

319 content_type: str | None = None, 

320 ) -> None: 

321 allow = ",".join(sorted(allowed_methods)) 

322 super().__init__( 

323 headers=headers, reason=reason, text=text, content_type=content_type 

324 ) 

325 self.headers["Allow"] = allow 

326 self._allowed: set[str] = set(allowed_methods) 

327 self._method = method 

328 

329 @property 

330 def allowed_methods(self) -> set[str]: 

331 return self._allowed 

332 

333 @property 

334 def method(self) -> str: 

335 return self._method 

336 

337 

338class HTTPNotAcceptable(HTTPClientError): 

339 status_code = 406 

340 

341 

342class HTTPProxyAuthenticationRequired(HTTPClientError): 

343 status_code = 407 

344 

345 

346class HTTPRequestTimeout(HTTPClientError): 

347 status_code = 408 

348 

349 

350class HTTPConflict(HTTPClientError): 

351 status_code = 409 

352 

353 

354class HTTPGone(HTTPClientError): 

355 status_code = 410 

356 

357 

358class HTTPLengthRequired(HTTPClientError): 

359 status_code = 411 

360 

361 

362class HTTPPreconditionFailed(HTTPClientError): 

363 status_code = 412 

364 

365 

366class HTTPRequestEntityTooLarge(HTTPClientError): 

367 status_code = 413 

368 

369 def __init__(self, max_size: int, actual_size: int, **kwargs: Any) -> None: 

370 kwargs.setdefault( 

371 "text", 

372 f"Maximum request body size {max_size} exceeded, " 

373 f"actual body size {actual_size}", 

374 ) 

375 super().__init__(**kwargs) 

376 

377 

378class HTTPRequestURITooLong(HTTPClientError): 

379 status_code = 414 

380 

381 

382class HTTPUnsupportedMediaType(HTTPClientError): 

383 status_code = 415 

384 

385 

386class HTTPRequestRangeNotSatisfiable(HTTPClientError): 

387 status_code = 416 

388 

389 

390class HTTPExpectationFailed(HTTPClientError): 

391 status_code = 417 

392 

393 

394class HTTPMisdirectedRequest(HTTPClientError): 

395 status_code = 421 

396 

397 

398class HTTPUnprocessableEntity(HTTPClientError): 

399 status_code = 422 

400 

401 

402class HTTPFailedDependency(HTTPClientError): 

403 status_code = 424 

404 

405 

406class HTTPUpgradeRequired(HTTPClientError): 

407 status_code = 426 

408 

409 

410class HTTPPreconditionRequired(HTTPClientError): 

411 status_code = 428 

412 

413 

414class HTTPTooManyRequests(HTTPClientError): 

415 status_code = 429 

416 

417 

418class HTTPRequestHeaderFieldsTooLarge(HTTPClientError): 

419 status_code = 431 

420 

421 

422class HTTPUnavailableForLegalReasons(HTTPClientError): 

423 status_code = 451 

424 

425 def __init__( 

426 self, 

427 link: StrOrURL | None, 

428 *, 

429 headers: LooseHeaders | None = None, 

430 reason: str | None = None, 

431 text: str | None = None, 

432 content_type: str | None = None, 

433 ) -> None: 

434 super().__init__( 

435 headers=headers, reason=reason, text=text, content_type=content_type 

436 ) 

437 self._link = None 

438 if link: 

439 self._link = URL(link) 

440 self.headers["Link"] = f'<{str(self._link)}>; rel="blocked-by"' 

441 

442 @property 

443 def link(self) -> URL | None: 

444 return self._link 

445 

446 

447############################################################ 

448# 5xx Server Error 

449############################################################ 

450# Response status codes beginning with the digit "5" indicate cases in 

451# which the server is aware that it has erred or is incapable of 

452# performing the request. Except when responding to a HEAD request, the 

453# server SHOULD include an entity containing an explanation of the error 

454# situation, and whether it is a temporary or permanent condition. User 

455# agents SHOULD display any included entity to the user. These response 

456# codes are applicable to any request method. 

457 

458 

459class HTTPServerError(HTTPError): 

460 pass 

461 

462 

463class HTTPInternalServerError(HTTPServerError): 

464 status_code = 500 

465 

466 

467class HTTPNotImplemented(HTTPServerError): 

468 status_code = 501 

469 

470 

471class HTTPBadGateway(HTTPServerError): 

472 status_code = 502 

473 

474 

475class HTTPServiceUnavailable(HTTPServerError): 

476 status_code = 503 

477 

478 

479class HTTPGatewayTimeout(HTTPServerError): 

480 status_code = 504 

481 

482 

483class HTTPVersionNotSupported(HTTPServerError): 

484 status_code = 505 

485 

486 

487class HTTPVariantAlsoNegotiates(HTTPServerError): 

488 status_code = 506 

489 

490 

491class HTTPInsufficientStorage(HTTPServerError): 

492 status_code = 507 

493 

494 

495class HTTPNotExtended(HTTPServerError): 

496 status_code = 510 

497 

498 

499class HTTPNetworkAuthenticationRequired(HTTPServerError): 

500 status_code = 511 

501 

502 

503def _initialize_default_reason() -> None: 

504 for obj in globals().values(): 

505 if isinstance(obj, type) and issubclass(obj, HTTPException): 

506 if obj.status_code >= 0: 

507 try: 

508 status = HTTPStatus(obj.status_code) 

509 obj.default_reason = status.phrase 

510 except ValueError: 

511 pass 

512 

513 

514_initialize_default_reason() 

515del _initialize_default_reason