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

1import warnings 

2from http import HTTPStatus 

3from typing import Any, Iterable, Optional, Set, Tuple 

4 

5from multidict import CIMultiDict 

6from yarl import URL 

7 

8from . import hdrs 

9from .helpers import CookieMixin 

10from .typedefs import LooseHeaders, StrOrURL 

11 

12__all__ = ( 

13 "HTTPException", 

14 "HTTPError", 

15 "HTTPRedirection", 

16 "HTTPSuccessful", 

17 "HTTPOk", 

18 "HTTPCreated", 

19 "HTTPAccepted", 

20 "HTTPNonAuthoritativeInformation", 

21 "HTTPNoContent", 

22 "HTTPResetContent", 

23 "HTTPPartialContent", 

24 "HTTPMove", 

25 "HTTPMultipleChoices", 

26 "HTTPMovedPermanently", 

27 "HTTPFound", 

28 "HTTPSeeOther", 

29 "HTTPNotModified", 

30 "HTTPUseProxy", 

31 "HTTPTemporaryRedirect", 

32 "HTTPPermanentRedirect", 

33 "HTTPClientError", 

34 "HTTPBadRequest", 

35 "HTTPUnauthorized", 

36 "HTTPPaymentRequired", 

37 "HTTPForbidden", 

38 "HTTPNotFound", 

39 "HTTPMethodNotAllowed", 

40 "HTTPNotAcceptable", 

41 "HTTPProxyAuthenticationRequired", 

42 "HTTPRequestTimeout", 

43 "HTTPConflict", 

44 "HTTPGone", 

45 "HTTPLengthRequired", 

46 "HTTPPreconditionFailed", 

47 "HTTPRequestEntityTooLarge", 

48 "HTTPRequestURITooLong", 

49 "HTTPUnsupportedMediaType", 

50 "HTTPRequestRangeNotSatisfiable", 

51 "HTTPExpectationFailed", 

52 "HTTPMisdirectedRequest", 

53 "HTTPUnprocessableEntity", 

54 "HTTPFailedDependency", 

55 "HTTPUpgradeRequired", 

56 "HTTPPreconditionRequired", 

57 "HTTPTooManyRequests", 

58 "HTTPRequestHeaderFieldsTooLarge", 

59 "HTTPUnavailableForLegalReasons", 

60 "HTTPServerError", 

61 "HTTPInternalServerError", 

62 "HTTPNotImplemented", 

63 "HTTPBadGateway", 

64 "HTTPServiceUnavailable", 

65 "HTTPGatewayTimeout", 

66 "HTTPVersionNotSupported", 

67 "HTTPVariantAlsoNegotiates", 

68 "HTTPInsufficientStorage", 

69 "HTTPNotExtended", 

70 "HTTPNetworkAuthenticationRequired", 

71) 

72 

73 

74class NotAppKeyWarning(UserWarning): 

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

76 

77 

78############################################################ 

79# HTTP Exceptions 

80############################################################ 

81 

82 

83class HTTPException(CookieMixin, Exception): 

84 # You should set in subclasses: 

85 # status = 200 

86 

87 status_code = -1 

88 empty_body = False 

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

90 

91 def __init__( 

92 self, 

93 *, 

94 headers: Optional[LooseHeaders] = None, 

95 reason: Optional[str] = None, 

96 text: Optional[str] = None, 

97 content_type: Optional[str] = None, 

98 ) -> None: 

99 if reason is None: 

100 reason = self.default_reason 

101 elif "\n" in reason: 

102 raise ValueError("Reason cannot contain \\n") 

103 

104 if text is None: 

105 if not self.empty_body: 

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

107 else: 

108 if self.empty_body: 

109 warnings.warn( 

110 "text argument is deprecated for HTTP status {} " 

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

112 "the response should be provided without a body".format( 

113 self.status_code 

114 ), 

115 DeprecationWarning, 

116 stacklevel=2, 

117 ) 

118 

119 if headers is not None: 

120 real_headers = CIMultiDict(headers) 

121 else: 

122 real_headers = CIMultiDict() 

123 

124 if content_type is not None: 

125 if not text: 

126 warnings.warn( 

127 "content_type without text is deprecated " 

128 "since 4.0 and scheduled for removal in 5.0 " 

129 "(#3462)", 

130 DeprecationWarning, 

131 stacklevel=2, 

132 ) 

133 real_headers[hdrs.CONTENT_TYPE] = content_type 

134 elif hdrs.CONTENT_TYPE not in real_headers and text: 

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

136 

137 self._reason = reason 

138 self._text = text 

139 self._headers = real_headers 

140 self.args = () 

141 

142 def __bool__(self) -> bool: 

143 return True 

144 

145 @property 

146 def status(self) -> int: 

147 return self.status_code 

148 

149 @property 

150 def reason(self) -> str: 

151 return self._reason 

152 

153 @property 

154 def text(self) -> Optional[str]: 

155 return self._text 

156 

157 @property 

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

159 return self._headers 

160 

161 def __str__(self) -> str: 

162 return self.reason 

163 

164 def __repr__(self) -> str: 

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

166 

167 __reduce__ = object.__reduce__ 

168 

169 def __getnewargs__(self) -> Tuple[Any, ...]: 

170 return self.args 

171 

172 

173class HTTPError(HTTPException): 

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

175 

176 

177class HTTPRedirection(HTTPException): 

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

179 

180 

181class HTTPSuccessful(HTTPException): 

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

183 

184 

185class HTTPOk(HTTPSuccessful): 

186 status_code = 200 

187 

188 

189class HTTPCreated(HTTPSuccessful): 

190 status_code = 201 

191 

192 

193class HTTPAccepted(HTTPSuccessful): 

194 status_code = 202 

195 

196 

197class HTTPNonAuthoritativeInformation(HTTPSuccessful): 

198 status_code = 203 

199 

200 

201class HTTPNoContent(HTTPSuccessful): 

202 status_code = 204 

203 empty_body = True 

204 

205 

206class HTTPResetContent(HTTPSuccessful): 

207 status_code = 205 

208 empty_body = True 

209 

210 

211class HTTPPartialContent(HTTPSuccessful): 

212 status_code = 206 

213 

214 

215############################################################ 

216# 3xx redirection 

217############################################################ 

218 

219 

220class HTTPMove(HTTPRedirection): 

221 def __init__( 

222 self, 

223 location: StrOrURL, 

224 *, 

225 headers: Optional[LooseHeaders] = None, 

226 reason: Optional[str] = None, 

227 text: Optional[str] = None, 

228 content_type: Optional[str] = None, 

229 ) -> None: 

230 if not location: 

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

232 super().__init__( 

233 headers=headers, reason=reason, text=text, content_type=content_type 

234 ) 

235 self._location = URL(location) 

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

237 

238 @property 

239 def location(self) -> URL: 

240 return self._location 

241 

242 

243class HTTPMultipleChoices(HTTPMove): 

244 status_code = 300 

245 

246 

247class HTTPMovedPermanently(HTTPMove): 

248 status_code = 301 

249 

250 

251class HTTPFound(HTTPMove): 

252 status_code = 302 

253 

254 

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

256# retrieved with GET): 

257class HTTPSeeOther(HTTPMove): 

258 status_code = 303 

259 

260 

261class HTTPNotModified(HTTPRedirection): 

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

263 status_code = 304 

264 empty_body = True 

265 

266 

267class HTTPUseProxy(HTTPMove): 

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

269 status_code = 305 

270 

271 

272class HTTPTemporaryRedirect(HTTPMove): 

273 status_code = 307 

274 

275 

276class HTTPPermanentRedirect(HTTPMove): 

277 status_code = 308 

278 

279 

280############################################################ 

281# 4xx client error 

282############################################################ 

283 

284 

285class HTTPClientError(HTTPError): 

286 pass 

287 

288 

289class HTTPBadRequest(HTTPClientError): 

290 status_code = 400 

291 

292 

293class HTTPUnauthorized(HTTPClientError): 

294 status_code = 401 

295 

296 

297class HTTPPaymentRequired(HTTPClientError): 

298 status_code = 402 

299 

300 

301class HTTPForbidden(HTTPClientError): 

302 status_code = 403 

303 

304 

305class HTTPNotFound(HTTPClientError): 

306 status_code = 404 

307 

308 

309class HTTPMethodNotAllowed(HTTPClientError): 

310 status_code = 405 

311 

312 def __init__( 

313 self, 

314 method: str, 

315 allowed_methods: Iterable[str], 

316 *, 

317 headers: Optional[LooseHeaders] = None, 

318 reason: Optional[str] = None, 

319 text: Optional[str] = None, 

320 content_type: Optional[str] = None, 

321 ) -> None: 

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

323 super().__init__( 

324 headers=headers, reason=reason, text=text, content_type=content_type 

325 ) 

326 self.headers["Allow"] = allow 

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

328 self._method = method 

329 

330 @property 

331 def allowed_methods(self) -> Set[str]: 

332 return self._allowed 

333 

334 @property 

335 def method(self) -> str: 

336 return self._method 

337 

338 

339class HTTPNotAcceptable(HTTPClientError): 

340 status_code = 406 

341 

342 

343class HTTPProxyAuthenticationRequired(HTTPClientError): 

344 status_code = 407 

345 

346 

347class HTTPRequestTimeout(HTTPClientError): 

348 status_code = 408 

349 

350 

351class HTTPConflict(HTTPClientError): 

352 status_code = 409 

353 

354 

355class HTTPGone(HTTPClientError): 

356 status_code = 410 

357 

358 

359class HTTPLengthRequired(HTTPClientError): 

360 status_code = 411 

361 

362 

363class HTTPPreconditionFailed(HTTPClientError): 

364 status_code = 412 

365 

366 

367class HTTPRequestEntityTooLarge(HTTPClientError): 

368 status_code = 413 

369 

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

371 kwargs.setdefault( 

372 "text", 

373 "Maximum request body size {} exceeded, " 

374 "actual body size {}".format(max_size, actual_size), 

375 ) 

376 super().__init__(**kwargs) 

377 

378 

379class HTTPRequestURITooLong(HTTPClientError): 

380 status_code = 414 

381 

382 

383class HTTPUnsupportedMediaType(HTTPClientError): 

384 status_code = 415 

385 

386 

387class HTTPRequestRangeNotSatisfiable(HTTPClientError): 

388 status_code = 416 

389 

390 

391class HTTPExpectationFailed(HTTPClientError): 

392 status_code = 417 

393 

394 

395class HTTPMisdirectedRequest(HTTPClientError): 

396 status_code = 421 

397 

398 

399class HTTPUnprocessableEntity(HTTPClientError): 

400 status_code = 422 

401 

402 

403class HTTPFailedDependency(HTTPClientError): 

404 status_code = 424 

405 

406 

407class HTTPUpgradeRequired(HTTPClientError): 

408 status_code = 426 

409 

410 

411class HTTPPreconditionRequired(HTTPClientError): 

412 status_code = 428 

413 

414 

415class HTTPTooManyRequests(HTTPClientError): 

416 status_code = 429 

417 

418 

419class HTTPRequestHeaderFieldsTooLarge(HTTPClientError): 

420 status_code = 431 

421 

422 

423class HTTPUnavailableForLegalReasons(HTTPClientError): 

424 status_code = 451 

425 

426 def __init__( 

427 self, 

428 link: Optional[StrOrURL], 

429 *, 

430 headers: Optional[LooseHeaders] = None, 

431 reason: Optional[str] = None, 

432 text: Optional[str] = None, 

433 content_type: Optional[str] = None, 

434 ) -> None: 

435 super().__init__( 

436 headers=headers, reason=reason, text=text, content_type=content_type 

437 ) 

438 self._link = None 

439 if link: 

440 self._link = URL(link) 

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

442 

443 @property 

444 def link(self) -> Optional[URL]: 

445 return self._link 

446 

447 

448############################################################ 

449# 5xx Server Error 

450############################################################ 

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

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

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

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

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

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

457# codes are applicable to any request method. 

458 

459 

460class HTTPServerError(HTTPError): 

461 pass 

462 

463 

464class HTTPInternalServerError(HTTPServerError): 

465 status_code = 500 

466 

467 

468class HTTPNotImplemented(HTTPServerError): 

469 status_code = 501 

470 

471 

472class HTTPBadGateway(HTTPServerError): 

473 status_code = 502 

474 

475 

476class HTTPServiceUnavailable(HTTPServerError): 

477 status_code = 503 

478 

479 

480class HTTPGatewayTimeout(HTTPServerError): 

481 status_code = 504 

482 

483 

484class HTTPVersionNotSupported(HTTPServerError): 

485 status_code = 505 

486 

487 

488class HTTPVariantAlsoNegotiates(HTTPServerError): 

489 status_code = 506 

490 

491 

492class HTTPInsufficientStorage(HTTPServerError): 

493 status_code = 507 

494 

495 

496class HTTPNotExtended(HTTPServerError): 

497 status_code = 510 

498 

499 

500class HTTPNetworkAuthenticationRequired(HTTPServerError): 

501 status_code = 511 

502 

503 

504def _initialize_default_reason() -> None: 

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

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

507 if obj.status_code >= 0: 

508 try: 

509 status = HTTPStatus(obj.status_code) 

510 obj.default_reason = status.phrase 

511 except ValueError: 

512 pass 

513 

514 

515_initialize_default_reason() 

516del _initialize_default_reason