Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/web

1import asyncio

2import collections.abc

3import datetime

4import enum

5import json

6import math

7import time

8import warnings

9from concurrent.futures import Executor

10from http import HTTPStatus

11from typing import (

12 TYPE_CHECKING,

13 Any,

14 Dict,

15 Iterator,

16 MutableMapping,

17 Optional,

18 Union,

19 cast,

20)

22from multidict import CIMultiDict, istr

24from . import hdrs, payload

25from .abc import AbstractStreamWriter

26from .compression_utils import ZLibCompressor

27from .helpers import (

28 ETAG_ANY,

29 QUOTED_ETAG_RE,

30 CookieMixin,

31 ETag,

32 HeadersMixin,

33 parse_http_date,

34 populate_with_cookies,

35 rfc822_formatted_time,

36 sentinel,

37 validate_etag_value,

38)

39from .http import SERVER_SOFTWARE, HttpVersion10, HttpVersion11

40from .payload import Payload

41from .typedefs import JSONEncoder, LooseHeaders

43__all__ = ("ContentCoding", "StreamResponse", "Response", "json_response")

46if TYPE_CHECKING: # pragma: no cover

47 from .web_request import BaseRequest

49 BaseClass = MutableMapping[str, Any]

50else:

51 BaseClass = collections.abc.MutableMapping

54class ContentCoding(enum.Enum):

55 # The content codings that we have support for.

56 #

57 # Additional registered codings are listed at:

58 # https://www.iana.org/assignments/http-parameters/http-parameters.xhtml#content-coding

59 deflate = "deflate"

60 gzip = "gzip"

61 identity = "identity"

64############################################################

65# HTTP Response classes

66############################################################

69class StreamResponse(BaseClass, HeadersMixin, CookieMixin):

70 __slots__ = (

71 "_length_check",

72 "_body",

73 "_keep_alive",

74 "_chunked",

75 "_compression",

76 "_compression_force",

77 "_req",

78 "_payload_writer",

79 "_eof_sent",

80 "_body_length",

81 "_state",

82 "_headers",

83 "_status",

84 "_reason",

85 "_cookies",

86 "__weakref__",

87 )

89 def __init__(

90 self,

91 *,

92 status: int = 200,

93 reason: Optional[str] = None,

94 headers: Optional[LooseHeaders] = None,

95 ) -> None:

96 super().__init__()

97 self._length_check = True

98 self._body = None

99 self._keep_alive: Optional[bool] = None

100 self._chunked = False

101 self._compression = False

102 self._compression_force: Optional[ContentCoding] = None

103

104 self._req: Optional[BaseRequest] = None

105 self._payload_writer: Optional[AbstractStreamWriter] = None

106 self._eof_sent = False

107 self._body_length = 0

108 self._state: Dict[str, Any] = {}

109

110 if headers is not None:

111 self._headers: CIMultiDict[str] = CIMultiDict(headers)

112 else:

113 self._headers = CIMultiDict()

114

115 self.set_status(status, reason)

116

117 @property

118 def prepared(self) -> bool:

119 return self._payload_writer is not None

120

121 @property

122 def task(self) -> "Optional[asyncio.Task[None]]":

123 if self._req:

124 return self._req.task

125 else:

126 return None

127

128 @property

129 def status(self) -> int:

130 return self._status

131

132 @property

133 def chunked(self) -> bool:

134 return self._chunked

135

136 @property

137 def compression(self) -> bool:

138 return self._compression

139

140 @property

141 def reason(self) -> str:

142 return self._reason

143

144 def set_status(

145 self,

146 status: int,

147 reason: Optional[str] = None,

148 ) -> None:

149 assert not self.prepared, (

150 "Cannot change the response status code after " "the headers have been sent"

151 )

152 self._status = int(status)

153 if reason is None:

154 try:

155 reason = HTTPStatus(self._status).phrase

156 except ValueError:

157 reason = ""

158 self._reason = reason

159

160 @property

161 def keep_alive(self) -> Optional[bool]:

162 return self._keep_alive

163

164 def force_close(self) -> None:

165 self._keep_alive = False

166

167 @property

168 def body_length(self) -> int:

169 return self._body_length

170

171 def enable_chunked_encoding(self) -> None:

172 """Enables automatic chunked transfer encoding."""

173 self._chunked = True

174

175 if hdrs.CONTENT_LENGTH in self._headers:

176 raise RuntimeError(

177 "You can't enable chunked encoding when " "a content length is set"

178 )

179

180 def enable_compression(self, force: Optional[ContentCoding] = None) -> None:

181 """Enables response compression encoding."""

182 # Backwards compatibility for when force was a bool <0.17.

183 self._compression = True

184 self._compression_force = force

185

186 @property

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

188 return self._headers

189

190 @property

191 def content_length(self) -> Optional[int]:

192 # Just a placeholder for adding setter

193 return super().content_length

194

195 @content_length.setter

196 def content_length(self, value: Optional[int]) -> None:

197 if value is not None:

198 value = int(value)

199 if self._chunked:

200 raise RuntimeError(

201 "You can't set content length when " "chunked encoding is enable"

202 )

203 self._headers[hdrs.CONTENT_LENGTH] = str(value)

204 else:

205 self._headers.pop(hdrs.CONTENT_LENGTH, None)

206

207 @property

208 def content_type(self) -> str:

209 # Just a placeholder for adding setter

210 return super().content_type

211

212 @content_type.setter

213 def content_type(self, value: str) -> None:

214 self.content_type # read header values if needed

215 self._content_type = str(value)

216 self._generate_content_type_header()

217

218 @property

219 def charset(self) -> Optional[str]:

220 # Just a placeholder for adding setter

221 return super().charset

222

223 @charset.setter

224 def charset(self, value: Optional[str]) -> None:

225 ctype = self.content_type # read header values if needed

226 if ctype == "application/octet-stream":

227 raise RuntimeError(

228 "Setting charset for application/octet-stream "

229 "doesn't make sense, setup content_type first"

230 )

231 assert self._content_dict is not None

232 if value is None:

233 self._content_dict.pop("charset", None)

234 else:

235 self._content_dict["charset"] = str(value).lower()

236 self._generate_content_type_header()

237

238 @property

239 def last_modified(self) -> Optional[datetime.datetime]:

240 """The value of Last-Modified HTTP header, or None.

241

242 This header is represented as a `datetime` object.

243 """

244 return parse_http_date(self._headers.get(hdrs.LAST_MODIFIED))

245

246 @last_modified.setter

247 def last_modified(

248 self, value: Optional[Union[int, float, datetime.datetime, str]]

249 ) -> None:

250 if value is None:

251 self._headers.pop(hdrs.LAST_MODIFIED, None)

252 elif isinstance(value, (int, float)):

253 self._headers[hdrs.LAST_MODIFIED] = time.strftime(

254 "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(math.ceil(value))

255 )

256 elif isinstance(value, datetime.datetime):

257 self._headers[hdrs.LAST_MODIFIED] = time.strftime(

258 "%a, %d %b %Y %H:%M:%S GMT", value.utctimetuple()

259 )

260 elif isinstance(value, str):

261 self._headers[hdrs.LAST_MODIFIED] = value

262

263 @property

264 def etag(self) -> Optional[ETag]:

265 quoted_value = self._headers.get(hdrs.ETAG)

266 if not quoted_value:

267 return None

268 elif quoted_value == ETAG_ANY:

269 return ETag(value=ETAG_ANY)

270 match = QUOTED_ETAG_RE.fullmatch(quoted_value)

271 if not match:

272 return None

273 is_weak, value = match.group(1, 2)

274 return ETag(

275 is_weak=bool(is_weak),

276 value=value,

277 )

278

279 @etag.setter

280 def etag(self, value: Optional[Union[ETag, str]]) -> None:

281 if value is None:

282 self._headers.pop(hdrs.ETAG, None)

283 elif (isinstance(value, str) and value == ETAG_ANY) or (

284 isinstance(value, ETag) and value.value == ETAG_ANY

285 ):

286 self._headers[hdrs.ETAG] = ETAG_ANY

287 elif isinstance(value, str):

288 validate_etag_value(value)

289 self._headers[hdrs.ETAG] = f'"{value}"'

290 elif isinstance(value, ETag) and isinstance(value.value, str): # type: ignore[redundant-expr]

291 validate_etag_value(value.value)

292 hdr_value = f'W/"{value.value}"' if value.is_weak else f'"{value.value}"'

293 self._headers[hdrs.ETAG] = hdr_value

294 else:

295 raise ValueError(

296 f"Unsupported etag type: {type(value)}. "

297 f"etag must be str, ETag or None"

298 )

299

300 def _generate_content_type_header(

301 self, CONTENT_TYPE: istr = hdrs.CONTENT_TYPE

302 ) -> None:

303 assert self._content_dict is not None

304 assert self._content_type is not None

305 params = "; ".join(f"{k}={v}" for k, v in self._content_dict.items())

306 if params:

307 ctype = self._content_type + "; " + params

308 else:

309 ctype = self._content_type

310 self._headers[CONTENT_TYPE] = ctype

311

312 async def _do_start_compression(self, coding: ContentCoding) -> None:

313 if coding != ContentCoding.identity:

314 assert self._payload_writer is not None

315 self._headers[hdrs.CONTENT_ENCODING] = coding.value

316 self._payload_writer.enable_compression(coding.value)

317 # Compressed payload may have different content length,

318 # remove the header

319 self._headers.popall(hdrs.CONTENT_LENGTH, None)

320

321 async def _start_compression(self, request: "BaseRequest") -> None:

322 if self._compression_force:

323 await self._do_start_compression(self._compression_force)

324 else:

325 accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()

326 for coding in ContentCoding:

327 if coding.value in accept_encoding:

328 await self._do_start_compression(coding)

329 return

330

331 async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:

332 if self._eof_sent:

333 return None

334 if self._payload_writer is not None:

335 return self._payload_writer

336

337 return await self._start(request)

338

339 async def _start(self, request: "BaseRequest") -> AbstractStreamWriter:

340 self._req = request

341 writer = self._payload_writer = request._payload_writer

342

343 await self._prepare_headers()

344 await request._prepare_hook(self)

345 await self._write_headers()

346

347 return writer

348

349 async def _prepare_headers(self) -> None:

350 request = self._req

351 assert request is not None

352 writer = self._payload_writer

353 assert writer is not None

354 keep_alive = self._keep_alive

355 if keep_alive is None:

356 keep_alive = request.keep_alive

357 self._keep_alive = keep_alive

358

359 version = request.version

360

361 headers = self._headers

362 populate_with_cookies(headers, self.cookies)

363

364 if self._compression:

365 await self._start_compression(request)

366

367 if self._chunked:

368 if version != HttpVersion11:

369 raise RuntimeError(

370 "Using chunked encoding is forbidden "

371 "for HTTP/{0.major}.{0.minor}".format(request.version)

372 )

373 writer.enable_chunking()

374 headers[hdrs.TRANSFER_ENCODING] = "chunked"

375 if hdrs.CONTENT_LENGTH in headers:

376 del headers[hdrs.CONTENT_LENGTH]

377 elif self._length_check:

378 writer.length = self.content_length

379 if writer.length is None:

380 if version >= HttpVersion11 and self.status != 204:

381 writer.enable_chunking()

382 headers[hdrs.TRANSFER_ENCODING] = "chunked"

383 if hdrs.CONTENT_LENGTH in headers:

384 del headers[hdrs.CONTENT_LENGTH]

385 else:

386 keep_alive = False

387 # HTTP 1.1: https://tools.ietf.org/html/rfc7230#section-3.3.2

388 # HTTP 1.0: https://tools.ietf.org/html/rfc1945#section-10.4

389 elif version >= HttpVersion11 and self.status in (100, 101, 102, 103, 204):

390 del headers[hdrs.CONTENT_LENGTH]

391

392 if self.status not in (204, 304):

393 headers.setdefault(hdrs.CONTENT_TYPE, "application/octet-stream")

394 headers.setdefault(hdrs.DATE, rfc822_formatted_time())

395 headers.setdefault(hdrs.SERVER, SERVER_SOFTWARE)

396

397 # connection header

398 if hdrs.CONNECTION not in headers:

399 if keep_alive:

400 if version == HttpVersion10:

401 headers[hdrs.CONNECTION] = "keep-alive"

402 else:

403 if version == HttpVersion11:

404 headers[hdrs.CONNECTION] = "close"

405

406 async def _write_headers(self) -> None:

407 request = self._req

408 assert request is not None

409 writer = self._payload_writer

410 assert writer is not None

411 # status line

412 version = request.version

413 status_line = "HTTP/{}.{} {} {}".format(

414 version[0], version[1], self._status, self._reason

415 )

416 await writer.write_headers(status_line, self._headers)

417

418 async def write(self, data: bytes) -> None:

419 assert isinstance(

420 data, (bytes, bytearray, memoryview)

421 ), "data argument must be byte-ish (%r)" % type(data)

422

423 if self._eof_sent:

424 raise RuntimeError("Cannot call write() after write_eof()")

425 if self._payload_writer is None:

426 raise RuntimeError("Cannot call write() before prepare()")

427

428 await self._payload_writer.write(data)

429

430 async def drain(self) -> None:

431 assert not self._eof_sent, "EOF has already been sent"

432 assert self._payload_writer is not None, "Response has not been started"

433 warnings.warn(

434 "drain method is deprecated, use await resp.write()",

435 DeprecationWarning,

436 stacklevel=2,

437 )

438 await self._payload_writer.drain()

439

440 async def write_eof(self, data: bytes = b"") -> None:

441 assert isinstance(

442 data, (bytes, bytearray, memoryview)

443 ), "data argument must be byte-ish (%r)" % type(data)

444

445 if self._eof_sent:

446 return

447

448 assert self._payload_writer is not None, "Response has not been started"

449

450 await self._payload_writer.write_eof(data)

451 self._eof_sent = True

452 self._req = None

453 self._body_length = self._payload_writer.output_size

454 self._payload_writer = None

455

456 def __repr__(self) -> str:

457 if self._eof_sent:

458 info = "eof"

459 elif self.prepared:

460 assert self._req is not None

461 info = f"{self._req.method} {self._req.path} "

462 else:

463 info = "not prepared"

464 return f"<{self.__class__.__name__} {self.reason} {info}>"

465

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

467 return self._state[key]

468

469 def __setitem__(self, key: str, value: Any) -> None:

470 self._state[key] = value

471

472 def __delitem__(self, key: str) -> None:

473 del self._state[key]

474

475 def __len__(self) -> int:

476 return len(self._state)

477

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

479 return iter(self._state)

480

481 def __hash__(self) -> int:

482 return hash(id(self))

483

484 def __eq__(self, other: object) -> bool:

485 return self is other

486

487

488class Response(StreamResponse):

489 __slots__ = (

490 "_body_payload",

491 "_compressed_body",

492 "_zlib_executor_size",

493 "_zlib_executor",

494 )

495

496 def __init__(

497 self,

498 *,

499 body: Any = None,

500 status: int = 200,

501 reason: Optional[str] = None,

502 text: Optional[str] = None,

503 headers: Optional[LooseHeaders] = None,

504 content_type: Optional[str] = None,

505 charset: Optional[str] = None,

506 zlib_executor_size: Optional[int] = None,

507 zlib_executor: Optional[Executor] = None,

508 ) -> None:

509 if body is not None and text is not None:

510 raise ValueError("body and text are not allowed together")

511

512 if headers is None:

513 real_headers: CIMultiDict[str] = CIMultiDict()

514 elif not isinstance(headers, CIMultiDict):

515 real_headers = CIMultiDict(headers)

516 else:

517 real_headers = headers # = cast('CIMultiDict[str]', headers)

518

519 if content_type is not None and "charset" in content_type:

520 raise ValueError("charset must not be in content_type " "argument")

521

522 if text is not None:

523 if hdrs.CONTENT_TYPE in real_headers:

524 if content_type or charset:

525 raise ValueError(

526 "passing both Content-Type header and "

527 "content_type or charset params "

528 "is forbidden"

529 )

530 else:

531 # fast path for filling headers

532 if not isinstance(text, str):

533 raise TypeError("text argument must be str (%r)" % type(text))

534 if content_type is None:

535 content_type = "text/plain"

536 if charset is None:

537 charset = "utf-8"

538 real_headers[hdrs.CONTENT_TYPE] = content_type + "; charset=" + charset

539 body = text.encode(charset)

540 text = None

541 else:

542 if hdrs.CONTENT_TYPE in real_headers:

543 if content_type is not None or charset is not None:

544 raise ValueError(

545 "passing both Content-Type header and "

546 "content_type or charset params "

547 "is forbidden"

548 )

549 else:

550 if content_type is not None:

551 if charset is not None:

552 content_type += "; charset=" + charset

553 real_headers[hdrs.CONTENT_TYPE] = content_type

554

555 super().__init__(status=status, reason=reason, headers=real_headers)

556

557 if text is not None:

558 self.text = text

559 else:

560 self.body = body

561

562 self._compressed_body: Optional[bytes] = None

563 self._zlib_executor_size = zlib_executor_size

564 self._zlib_executor = zlib_executor

565

566 @property

567 def body(self) -> Optional[Union[bytes, Payload]]:

568 return self._body

569

570 @body.setter

571 def body(self, body: bytes) -> None:

572 if body is None:

573 self._body: Optional[bytes] = None

574 self._body_payload: bool = False

575 elif isinstance(body, (bytes, bytearray)):

576 self._body = body

577 self._body_payload = False

578 else:

579 try:

580 self._body = body = payload.PAYLOAD_REGISTRY.get(body)

581 except payload.LookupError:

582 raise ValueError("Unsupported body type %r" % type(body))

583

584 self._body_payload = True

585

586 headers = self._headers

587

588 # set content-type

589 if hdrs.CONTENT_TYPE not in headers:

590 headers[hdrs.CONTENT_TYPE] = body.content_type

591

592 # copy payload headers

593 if body.headers:

594 for key, value in body.headers.items():

595 if key not in headers:

596 headers[key] = value

597

598 self._compressed_body = None

599

600 @property

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

602 if self._body is None:

603 return None

604 return self._body.decode(self.charset or "utf-8")

605

606 @text.setter

607 def text(self, text: str) -> None:

608 assert isinstance(text, str), "text argument must be str (%r)" % type(text)

609

610 if self.content_type == "application/octet-stream":

611 self.content_type = "text/plain"

612 if self.charset is None:

613 self.charset = "utf-8"

614

615 self._body = text.encode(self.charset)

616 self._body_payload = False

617 self._compressed_body = None

618

619 @property

620 def content_length(self) -> Optional[int]:

621 if self._chunked:

622 return None

623

624 if hdrs.CONTENT_LENGTH in self._headers:

625 return super().content_length

626

627 if self._compressed_body is not None:

628 # Return length of the compressed body

629 return len(self._compressed_body)

630 elif self._body_payload:

631 # A payload without content length, or a compressed payload

632 return None

633 elif self._body is not None:

634 return len(self._body)

635 else:

636 return 0

637

638 @content_length.setter

639 def content_length(self, value: Optional[int]) -> None:

640 raise RuntimeError("Content length is set automatically")

641

642 async def write_eof(self, data: bytes = b"") -> None:

643 if self._eof_sent:

644 return

645 if self._compressed_body is None:

646 body: Optional[Union[bytes, Payload]] = self._body

647 else:

648 body = self._compressed_body

649 assert not data, f"data arg is not supported, got {data!r}"

650 assert self._req is not None

651 assert self._payload_writer is not None

652 if body is not None:

653 if self._req._method == hdrs.METH_HEAD or self._status in [204, 304]:

654 await super().write_eof()

655 elif self._body_payload:

656 payload = cast(Payload, body)

657 await payload.write(self._payload_writer)

658 await super().write_eof()

659 else:

660 await super().write_eof(cast(bytes, body))

661 else:

662 await super().write_eof()

663

664 async def _start(self, request: "BaseRequest") -> AbstractStreamWriter:

665 if not self._chunked and hdrs.CONTENT_LENGTH not in self._headers:

666 if self._body_payload:

667 size = cast(Payload, self._body).size

668 if size is not None:

669 self._headers[hdrs.CONTENT_LENGTH] = str(size)

670 else:

671 body_len = len(self._body) if self._body else "0"

672 self._headers[hdrs.CONTENT_LENGTH] = str(body_len)

673

674 return await super()._start(request)

675

676 async def _do_start_compression(self, coding: ContentCoding) -> None:

677 if self._body_payload or self._chunked:

678 return await super()._do_start_compression(coding)

679

680 if coding != ContentCoding.identity:

681 # Instead of using _payload_writer.enable_compression,

682 # compress the whole body

683 compressor = ZLibCompressor(

684 encoding=str(coding.value),

685 max_sync_chunk_size=self._zlib_executor_size,

686 executor=self._zlib_executor,

687 )

688 assert self._body is not None

689 if self._zlib_executor_size is None and len(self._body) > 1024 * 1024:

690 warnings.warn(

691 "Synchronous compression of large response bodies "

692 f"({len(self._body)} bytes) might block the async event loop. "

693 "Consider providing a custom value to zlib_executor_size/"

694 "zlib_executor response properties or disabling compression on it."

695 )

696 self._compressed_body = (

697 await compressor.compress(self._body) + compressor.flush()

698 )

699 assert self._compressed_body is not None

700

701 self._headers[hdrs.CONTENT_ENCODING] = coding.value

702 self._headers[hdrs.CONTENT_LENGTH] = str(len(self._compressed_body))

703

704

705def json_response(

706 data: Any = sentinel,

707 *,

708 text: Optional[str] = None,

709 body: Optional[bytes] = None,

710 status: int = 200,

711 reason: Optional[str] = None,

712 headers: Optional[LooseHeaders] = None,

713 content_type: str = "application/json",

714 dumps: JSONEncoder = json.dumps,

715) -> Response:

716 if data is not sentinel:

717 if text or body:

718 raise ValueError("only one of data, text, or body should be specified")

719 else:

720 text = dumps(data)

721 return Response(

722 text=text,

723 body=body,

724 status=status,

725 reason=reason,

726 headers=headers,

727 content_type=content_type,

728 )

Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/web_response.py: 25%

430 statements