codekingpro/portable-devtools
115k
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 http.cookies import SimpleCookie
12from typing import (
13 TYPE_CHECKING,
14 Any,
15 Dict,
16 Iterator,
17 MutableMapping,
18 Optional,
19 Union,
20 cast,
21)
22
23from multidict import CIMultiDict, istr
24
25from . import hdrs, payload
26from .abc import AbstractStreamWriter
27from .compression_utils import ZLibCompressor
28from .helpers import (
29 ETAG_ANY,
30 QUOTED_ETAG_RE,
31 ETag,
32 HeadersMixin,
33 must_be_empty_body,
34 parse_http_date,
35 rfc822_formatted_time,
36 sentinel,
37 should_remove_content_length,
38 validate_etag_value,
39)
40from .http import SERVER_SOFTWARE, HttpVersion10, HttpVersion11
41from .payload import Payload
42from .typedefs import JSONEncoder, LooseHeaders
43
44REASON_PHRASES = {http_status.value: http_status.phrase for http_status in HTTPStatus}
45LARGE_BODY_SIZE = 1024**2
46
47__all__ = ("ContentCoding", "StreamResponse", "Response", "json_response")
48
49
50if TYPE_CHECKING:
51 from .web_request import BaseRequest
52
53 BaseClass = MutableMapping[str, Any]
54else:
55 BaseClass = collections.abc.MutableMapping
56
57
58# TODO(py311): Convert to StrEnum for wider use
59class ContentCoding(enum.Enum):
60 # The content codings that we have support for.
61 #
62 # Additional registered codings are listed at:
63 # https://www.iana.org/assignments/http-parameters/http-parameters.xhtml#content-coding
64 deflate = "deflate"
65 gzip = "gzip"
66 identity = "identity"
67
68
69CONTENT_CODINGS = {coding.value: coding for coding in ContentCoding}
70
71############################################################
72# HTTP Response classes
73############################################################
74
75
76class StreamResponse(BaseClass, HeadersMixin):
77
78 _body: Union[None, bytes, bytearray, Payload]
79 _length_check = True
80 _body = None
81 _keep_alive: Optional[bool] = None
82 _chunked: bool = False
83 _compression: bool = False
84 _compression_strategy: Optional[int] = None
85 _compression_force: Optional[ContentCoding] = None
86 _req: Optional["BaseRequest"] = None
87 _payload_writer: Optional[AbstractStreamWriter] = None
88 _eof_sent: bool = False
89 _must_be_empty_body: Optional[bool] = None
90 _body_length = 0
91 _cookies: Optional[SimpleCookie] = None
92 _send_headers_immediately = True
93
94 def __init__(
95 self,
96 *,
97 status: int = 200,
98 reason: Optional[str] = None,
99 headers: Optional[LooseHeaders] = None,
100 _real_headers: Optional[CIMultiDict[str]] = None,
101 ) -> None:
102 """Initialize a new stream response object.
103
104 _real_headers is an internal parameter used to pass a pre-populated
105 headers object. It is used by the `Response` class to avoid copying
106 the headers when creating a new response object. It is not intended
107 to be used by external code.
108 """
109 self._state: Dict[str, Any] = {}
110
111 if _real_headers is not None:
112 self._headers = _real_headers
113 elif headers is not None:
114 self._headers: CIMultiDict[str] = CIMultiDict(headers)
115 else:
116 self._headers = CIMultiDict()
117
118 self._set_status(status, reason)
119
120 @property
121 def prepared(self) -> bool:
122 return self._eof_sent or self._payload_writer is not None
123
124 @property
125 def task(self) -> "Optional[asyncio.Task[None]]":
126 if self._req:
127 return self._req.task
128 else:
129 return None
130
131 @property
132 def status(self) -> int:
133 return self._status
134
135 @property
136 def chunked(self) -> bool:
137 return self._chunked
138
139 @property
140 def compression(self) -> bool:
141 return self._compression
142
143 @property
144 def reason(self) -> str:
145 return self._reason
146
147 def set_status(
148 self,
149 status: int,
150 reason: Optional[str] = None,
151 ) -> None:
152 assert (
153 not self.prepared
154 ), "Cannot change the response status code after the headers have been sent"
155 self._set_status(status, reason)
156
157 def _set_status(self, status: int, reason: Optional[str]) -> None:
158 self._status = int(status)
159 if reason is None:
160 reason = REASON_PHRASES.get(self._status, "")
161 elif "\r" in reason or "\n" in reason:
162 raise ValueError("Reason cannot contain \\r or \\n")
163 self._reason = reason
164
165 @property
166 def keep_alive(self) -> Optional[bool]:
167 return self._keep_alive
168
169 def force_close(self) -> None:
170 self._keep_alive = False
171
172 @property
173 def body_length(self) -> int:
174 return self._body_length
175
176 @property
177 def output_length(self) -> int:
178 warnings.warn("output_length is deprecated", DeprecationWarning)
179 assert self._payload_writer
180 return self._payload_writer.buffer_size
181
182 def enable_chunked_encoding(self, chunk_size: Optional[int] = None) -> None:
183 """Enables automatic chunked transfer encoding."""
184 if hdrs.CONTENT_LENGTH in self._headers:
185 raise RuntimeError(
186 "You can't enable chunked encoding when a content length is set"
187 )
188 if chunk_size is not None:
189 warnings.warn("Chunk size is deprecated #1615", DeprecationWarning)
190 self._chunked = True
191
192 def enable_compression(
193 self,
194 force: Optional[Union[bool, ContentCoding]] = None,
195 strategy: Optional[int] = None,
196 ) -> None:
197 """Enables response compression encoding."""
198 # Backwards compatibility for when force was a bool <0.17.
199 if isinstance(force, bool):
200 force = ContentCoding.deflate if force else ContentCoding.identity
201 warnings.warn(
202 "Using boolean for force is deprecated #3318", DeprecationWarning
203 )
204 elif force is not None:
205 assert isinstance(
206 force, ContentCoding
207 ), "force should one of None, bool or ContentEncoding"
208
209 self._compression = True
210 self._compression_force = force
211 self._compression_strategy = strategy
212
213 @property
214 def headers(self) -> "CIMultiDict[str]":
215 return self._headers
216
217 @property
218 def cookies(self) -> SimpleCookie:
219 if self._cookies is None:
220 self._cookies = SimpleCookie()
221 return self._cookies
222
223 def set_cookie(
224 self,
225 name: str,
226 value: str,
227 *,
228 expires: Optional[str] = None,
229 domain: Optional[str] = None,
230 max_age: Optional[Union[int, str]] = None,
231 path: str = "/",
232 secure: Optional[bool] = None,
233 httponly: Optional[bool] = None,
234 version: Optional[str] = None,
235 samesite: Optional[str] = None,
236 partitioned: Optional[bool] = None,
237 ) -> None:
238 """Set or update response cookie.
239
240 Sets new cookie or updates existent with new value.
241 Also updates only those params which are not None.
242 """
243 if self._cookies is None:
244 self._cookies = SimpleCookie()
245
246 self._cookies[name] = value
247 c = self._cookies[name]
248
249 if expires is not None:
250 c["expires"] = expires
251 elif c.get("expires") == "Thu, 01 Jan 1970 00:00:00 GMT":
252 del c["expires"]
253
254 if domain is not None:
255 c["domain"] = domain
256
257 if max_age is not None:
258 c["max-age"] = str(max_age)
259 elif "max-age" in c:
260 del c["max-age"]
261
262 c["path"] = path
263
264 if secure is not None:
265 c["secure"] = secure
266 if httponly is not None:
267 c["httponly"] = httponly
268 if version is not None:
269 c["version"] = version
270 if samesite is not None:
271 c["samesite"] = samesite
272
273 if partitioned is not None:
274 c["partitioned"] = partitioned
275
276 def del_cookie(
277 self,
278 name: str,
279 *,
280 domain: Optional[str] = None,
281 path: str = "/",
282 secure: Optional[bool] = None,
283 httponly: Optional[bool] = None,
284 samesite: Optional[str] = None,
285 ) -> None:
286 """Delete cookie.
287
288 Creates new empty expired cookie.
289 """
290 # TODO: do we need domain/path here?
291 if self._cookies is not None:
292 self._cookies.pop(name, None)
293 self.set_cookie(
294 name,
295 "",
296 max_age=0,
297 expires="Thu, 01 Jan 1970 00:00:00 GMT",
298 domain=domain,
299 path=path,
300 secure=secure,
301 httponly=httponly,
302 samesite=samesite,
303 )
304
305 @property
306 def content_length(self) -> Optional[int]:
307 # Just a placeholder for adding setter
308 return super().content_length
309
310 @content_length.setter
311 def content_length(self, value: Optional[int]) -> None:
312 if value is not None:
313 value = int(value)
314 if self._chunked:
315 raise RuntimeError(
316 "You can't set content length when chunked encoding is enable"
317 )
318 self._headers[hdrs.CONTENT_LENGTH] = str(value)
319 else:
320 self._headers.pop(hdrs.CONTENT_LENGTH, None)
321
322 @property
323 def content_type(self) -> str:
324 # Just a placeholder for adding setter
325 return super().content_type
326
327 @content_type.setter
328 def content_type(self, value: str) -> None:
329 self.content_type # read header values if needed
330 self._content_type = str(value)
331 self._generate_content_type_header()
332
333 @property
334 def charset(self) -> Optional[str]:
335 # Just a placeholder for adding setter
336 return super().charset
337
338 @charset.setter
339 def charset(self, value: Optional[str]) -> None:
340 ctype = self.content_type # read header values if needed
341 if ctype == "application/octet-stream":
342 raise RuntimeError(
343 "Setting charset for application/octet-stream "
344 "doesn't make sense, setup content_type first"
345 )
346 assert self._content_dict is not None
347 if value is None:
348 self._content_dict.pop("charset", None)
349 else:
350 self._content_dict["charset"] = str(value).lower()
351 self._generate_content_type_header()
352
353 @property
354 def last_modified(self) -> Optional[datetime.datetime]:
355 """The value of Last-Modified HTTP header, or None.
356
357 This header is represented as a `datetime` object.
358 """
359 return parse_http_date(self._headers.get(hdrs.LAST_MODIFIED))
360
361 @last_modified.setter
362 def last_modified(
363 self, value: Optional[Union[int, float, datetime.datetime, str]]
364 ) -> None:
365 if value is None:
366 self._headers.pop(hdrs.LAST_MODIFIED, None)
367 elif isinstance(value, (int, float)):
368 self._headers[hdrs.LAST_MODIFIED] = time.strftime(
369 "%a, %d %b %Y %H:%M:%S GMT", time.gmtime(math.ceil(value))
370 )
371 elif isinstance(value, datetime.datetime):
372 self._headers[hdrs.LAST_MODIFIED] = time.strftime(
373 "%a, %d %b %Y %H:%M:%S GMT", value.utctimetuple()
374 )
375 elif isinstance(value, str):
376 self._headers[hdrs.LAST_MODIFIED] = value
377 else:
378 msg = f"Unsupported type for last_modified: {type(value).__name__}"
379 raise TypeError(msg)
380
381 @property
382 def etag(self) -> Optional[ETag]:
383 quoted_value = self._headers.get(hdrs.ETAG)
384 if not quoted_value:
385 return None
386 elif quoted_value == ETAG_ANY:
387 return ETag(value=ETAG_ANY)
388 match = QUOTED_ETAG_RE.fullmatch(quoted_value)
389 if not match:
390 return None
391 is_weak, value = match.group(1, 2)
392 return ETag(
393 is_weak=bool(is_weak),
394 value=value,
395 )
396
397 @etag.setter
398 def etag(self, value: Optional[Union[ETag, str]]) -> None:
399 if value is None:
400 self._headers.pop(hdrs.ETAG, None)
401 elif (isinstance(value, str) and value == ETAG_ANY) or (
402 isinstance(value, ETag) and value.value == ETAG_ANY
403 ):
404 self._headers[hdrs.ETAG] = ETAG_ANY
405 elif isinstance(value, str):
406 validate_etag_value(value)
407 self._headers[hdrs.ETAG] = f'"{value}"'
408 elif isinstance(value, ETag) and isinstance(value.value, str):
409 validate_etag_value(value.value)
410 hdr_value = f'W/"{value.value}"' if value.is_weak else f'"{value.value}"'
411 self._headers[hdrs.ETAG] = hdr_value
412 else:
413 raise ValueError(
414 f"Unsupported etag type: {type(value)}. "
415 f"etag must be str, ETag or None"
416 )
417
418 def _generate_content_type_header(
419 self, CONTENT_TYPE: istr = hdrs.CONTENT_TYPE
420 ) -> None:
421 assert self._content_dict is not None
422 assert self._content_type is not None
423 params = "; ".join(f"{k}={v}" for k, v in self._content_dict.items())
424 if params:
425 ctype = self._content_type + "; " + params
426 else:
427 ctype = self._content_type
428 self._headers[CONTENT_TYPE] = ctype
429
430 async def _do_start_compression(self, coding: ContentCoding) -> None:
431 if coding is ContentCoding.identity:
432 return
433 assert self._payload_writer is not None
434 self._headers[hdrs.CONTENT_ENCODING] = coding.value
435 self._payload_writer.enable_compression(
436 coding.value, self._compression_strategy
437 )
438 # Compressed payload may have different content length,
439 # remove the header
440 self._headers.popall(hdrs.CONTENT_LENGTH, None)
441
442 async def _start_compression(self, request: "BaseRequest") -> None:
443 if self._compression_force:
444 await self._do_start_compression(self._compression_force)
445 return
446 # Encoding comparisons should be case-insensitive
447 # https://www.rfc-editor.org/rfc/rfc9110#section-8.4.1
448 accept_encoding = request.headers.get(hdrs.ACCEPT_ENCODING, "").lower()
449 for value, coding in CONTENT_CODINGS.items():
450 if value in accept_encoding:
451 await self._do_start_compression(coding)
452 return
453
454 async def prepare(self, request: "BaseRequest") -> Optional[AbstractStreamWriter]:
455 if self._eof_sent:
456 return None
457 if self._payload_writer is not None:
458 return self._payload_writer
459 self._must_be_empty_body = must_be_empty_body(request.method, self.status)
460 return await self._start(request)
461
462 async def _start(self, request: "BaseRequest") -> AbstractStreamWriter:
463 self._req = request
464 writer = self._payload_writer = request._payload_writer
465
466 await self._prepare_headers()
467 await request._prepare_hook(self)
468 await self._write_headers()
469
470 return writer
471
472 async def _prepare_headers(self) -> None:
473 request = self._req
474 assert request is not None
475 writer = self._payload_writer
476 assert writer is not None
477 keep_alive = self._keep_alive
478 if keep_alive is None:
479 keep_alive = request.keep_alive
480 self._keep_alive = keep_alive
481
482 version = request.version
483
484 headers = self._headers
485 if self._cookies:
486 for cookie in self._cookies.values():
487 value = cookie.output(header="")[1:]
488 headers.add(hdrs.SET_COOKIE, value)
489
490 if self._compression:
491 await self._start_compression(request)
492
493 if self._chunked:
494 if version != HttpVersion11:
495 raise RuntimeError(
496 "Using chunked encoding is forbidden "
497 "for HTTP/{0.major}.{0.minor}".format(request.version)
498 )
499 if not self._must_be_empty_body:
500 writer.enable_chunking()
501 headers[hdrs.TRANSFER_ENCODING] = "chunked"
502 elif self._length_check: # Disabled for WebSockets
503 writer.length = self.content_length
504 if writer.length is None:
505 if version >= HttpVersion11:
506 if not self._must_be_empty_body:
507 writer.enable_chunking()
508 headers[hdrs.TRANSFER_ENCODING] = "chunked"
509 elif not self._must_be_empty_body:
510 keep_alive = False
511
512 # HTTP 1.1: https://tools.ietf.org/html/rfc7230#section-3.3.2
513 # HTTP 1.0: https://tools.ietf.org/html/rfc1945#section-10.4
514 if self._must_be_empty_body:
515 if hdrs.CONTENT_LENGTH in headers and should_remove_content_length(
516 request.method, self.status
517 ):
518 del headers[hdrs.CONTENT_LENGTH]
519 # https://datatracker.ietf.org/doc/html/rfc9112#section-6.1-10
520 # https://datatracker.ietf.org/doc/html/rfc9112#section-6.1-13
521 if hdrs.TRANSFER_ENCODING in headers:
522 del headers[hdrs.TRANSFER_ENCODING]
523 elif (writer.length if self._length_check else self.content_length) != 0:
524 # https://www.rfc-editor.org/rfc/rfc9110#section-8.3-5
525 headers.setdefault(hdrs.CONTENT_TYPE, "application/octet-stream")
526 headers.setdefault(hdrs.DATE, rfc822_formatted_time())
527 headers.setdefault(hdrs.SERVER, SERVER_SOFTWARE)
528
529 # connection header
530 if hdrs.CONNECTION not in headers:
531 if keep_alive:
532 if version == HttpVersion10:
533 headers[hdrs.CONNECTION] = "keep-alive"
534 elif version == HttpVersion11:
535 headers[hdrs.CONNECTION] = "close"
536
537 async def _write_headers(self) -> None:
538 request = self._req
539 assert request is not None
540 writer = self._payload_writer
541 assert writer is not None
542 # status line
543 version = request.version
544 status_line = f"HTTP/{version[0]}.{version[1]} {self._status} {self._reason}"
545 await writer.write_headers(status_line, self._headers)
546 # Send headers immediately if not opted into buffering
547 if self._send_headers_immediately:
548 writer.send_headers()
549
550 async def write(self, data: Union[bytes, bytearray, memoryview]) -> None:
551 assert isinstance(
552 data, (bytes, bytearray, memoryview)
553 ), "data argument must be byte-ish (%r)" % type(data)
554
555 if self._eof_sent:
556 raise RuntimeError("Cannot call write() after write_eof()")
557 if self._payload_writer is None:
558 raise RuntimeError("Cannot call write() before prepare()")
559
560 await self._payload_writer.write(data)
561
562 async def drain(self) -> None:
563 assert not self._eof_sent, "EOF has already been sent"
564 assert self._payload_writer is not None, "Response has not been started"
565 warnings.warn(
566 "drain method is deprecated, use await resp.write()",
567 DeprecationWarning,
568 stacklevel=2,
569 )
570 await self._payload_writer.drain()
571
572 async def write_eof(self, data: bytes = b"") -> None:
573 assert isinstance(
574 data, (bytes, bytearray, memoryview)
575 ), "data argument must be byte-ish (%r)" % type(data)
576
577 if self._eof_sent:
578 return
579
580 assert self._payload_writer is not None, "Response has not been started"
581
582 await self._payload_writer.write_eof(data)
583 self._eof_sent = True
584 self._req = None
585 self._body_length = self._payload_writer.output_size
586 self._payload_writer = None
587
588 def __repr__(self) -> str:
589 if self._eof_sent:
590 info = "eof"
591 elif self.prepared:
592 assert self._req is not None
593 info = f"{self._req.method} {self._req.path} "
594 else:
595 info = "not prepared"
596 return f"<{self.__class__.__name__} {self.reason} {info}>"
597
598 def __getitem__(self, key: str) -> Any:
599 return self._state[key]
600
601 def __setitem__(self, key: str, value: Any) -> None:
602 self._state[key] = value
603
604 def __delitem__(self, key: str) -> None:
605 del self._state[key]
606
607 def __len__(self) -> int:
608 return len(self._state)
609
610 def __iter__(self) -> Iterator[str]:
611 return iter(self._state)
612
613 def __hash__(self) -> int:
614 return hash(id(self))
615
616 def __eq__(self, other: object) -> bool:
617 return self is other
618
619 def __bool__(self) -> bool:
620 return True
621
622
623class Response(StreamResponse):
624
625 _compressed_body: Optional[bytes] = None
626 _send_headers_immediately = False
627
628 def __init__(
629 self,
630 *,
631 body: Any = None,
632 status: int = 200,
633 reason: Optional[str] = None,
634 text: Optional[str] = None,
635 headers: Optional[LooseHeaders] = None,
636 content_type: Optional[str] = None,
637 charset: Optional[str] = None,
638 zlib_executor_size: Optional[int] = None,
639 zlib_executor: Optional[Executor] = None,
640 ) -> None:
641 if body is not None and text is not None:
642 raise ValueError("body and text are not allowed together")
643
644 if headers is None:
645 real_headers: CIMultiDict[str] = CIMultiDict()
646 else:
647 real_headers = CIMultiDict(headers)
648
649 if content_type is not None and "charset" in content_type:
650 raise ValueError("charset must not be in content_type argument")
651
652 if text is not None:
653 if hdrs.CONTENT_TYPE in real_headers:
654 if content_type or charset:
655 raise ValueError(
656 "passing both Content-Type header and "
657 "content_type or charset params "
658 "is forbidden"
659 )
660 else:
661 # fast path for filling headers
662 if not isinstance(text, str):
663 raise TypeError("text argument must be str (%r)" % type(text))
664 if content_type is None:
665 content_type = "text/plain"
666 if charset is None:
667 charset = "utf-8"
668 real_headers[hdrs.CONTENT_TYPE] = content_type + "; charset=" + charset
669 body = text.encode(charset)
670 text = None
671 elif hdrs.CONTENT_TYPE in real_headers:
672 if content_type is not None or charset is not None:
673 raise ValueError(
674 "passing both Content-Type header and "
675 "content_type or charset params "
676 "is forbidden"
677 )
678 elif content_type is not None:
679 if charset is not None:
680 content_type += "; charset=" + charset
681 real_headers[hdrs.CONTENT_TYPE] = content_type
682
683 super().__init__(status=status, reason=reason, _real_headers=real_headers)
684
685 if text is not None:
686 self.text = text
687 else:
688 self.body = body
689
690 self._zlib_executor_size = zlib_executor_size
691 self._zlib_executor = zlib_executor
692
693 @property
694 def body(self) -> Optional[Union[bytes, Payload]]:
695 return self._body
696
697 @body.setter
698 def body(self, body: Any) -> None:
699 if body is None:
700 self._body = None
701 elif isinstance(body, (bytes, bytearray)):
702 self._body = body
703 else:
704 try:
705 self._body = body = payload.PAYLOAD_REGISTRY.get(body)
706 except payload.LookupError:
707 raise ValueError("Unsupported body type %r" % type(body))
708
709 headers = self._headers
710
711 # set content-type
712 if hdrs.CONTENT_TYPE not in headers:
713 headers[hdrs.CONTENT_TYPE] = body.content_type
714
715 # copy payload headers
716 if body.headers:
717 for key, value in body.headers.items():
718 if key not in headers:
719 headers[key] = value
720
721 self._compressed_body = None
722
723 @property
724 def text(self) -> Optional[str]:
725 if self._body is None:
726 return None
727 # Note: When _body is a Payload (e.g. FilePayload), this may do blocking I/O
728 # This is generally safe as most common payloads (BytesPayload, StringPayload)
729 # don't do blocking I/O, but be careful with file-based payloads
730 return self._body.decode(self.charset or "utf-8")
731
732 @text.setter
733 def text(self, text: str) -> None:
734 assert text is None or isinstance(
735 text, str
736 ), "text argument must be str (%r)" % type(text)
737
738 if self.content_type == "application/octet-stream":
739 self.content_type = "text/plain"
740 if self.charset is None:
741 self.charset = "utf-8"
742
743 self._body = text.encode(self.charset)
744 self._compressed_body = None
745
746 @property
747 def content_length(self) -> Optional[int]:
748 if self._chunked:
749 return None
750
751 if hdrs.CONTENT_LENGTH in self._headers:
752 return int(self._headers[hdrs.CONTENT_LENGTH])
753
754 if self._compressed_body is not None:
755 # Return length of the compressed body
756 return len(self._compressed_body)
757 elif isinstance(self._body, Payload):
758 # A payload without content length, or a compressed payload
759 return None
760 elif self._body is not None:
761 return len(self._body)
762 else:
763 return 0
764
765 @content_length.setter
766 def content_length(self, value: Optional[int]) -> None:
767 raise RuntimeError("Content length is set automatically")
768
769 async def write_eof(self, data: bytes = b"") -> None:
770 if self._eof_sent:
771 return
772 if self._compressed_body is None:
773 body: Optional[Union[bytes, Payload]] = self._body
774 else:
775 body = self._compressed_body
776 assert not data, f"data arg is not supported, got {data!r}"
777 assert self._req is not None
778 assert self._payload_writer is not None
779 if body is None or self._must_be_empty_body:
780 await super().write_eof()
781 elif isinstance(self._body, Payload):
782 await self._body.write(self._payload_writer)
783 await self._body.close()
784 await super().write_eof()
785 else:
786 await super().write_eof(cast(bytes, body))
787
788 async def _start(self, request: "BaseRequest") -> AbstractStreamWriter:
789 if hdrs.CONTENT_LENGTH in self._headers:
790 if should_remove_content_length(request.method, self.status):
791 del self._headers[hdrs.CONTENT_LENGTH]
792 elif not self._chunked:
793 if isinstance(self._body, Payload):
794 if (size := self._body.size) is not None:
795 self._headers[hdrs.CONTENT_LENGTH] = str(size)
796 else:
797 body_len = len(self._body) if self._body else "0"
798 # https://www.rfc-editor.org/rfc/rfc9110.html#section-8.6-7
799 if body_len != "0" or (
800 self.status != 304 and request.method not in hdrs.METH_HEAD_ALL
801 ):
802 self._headers[hdrs.CONTENT_LENGTH] = str(body_len)
803
804 return await super()._start(request)
805
806 async def _do_start_compression(self, coding: ContentCoding) -> None:
807 if self._chunked or isinstance(self._body, Payload):
808 return await super()._do_start_compression(coding)
809 if coding is ContentCoding.identity:
810 return
811 # Instead of using _payload_writer.enable_compression,
812 # compress the whole body
813 compressor = ZLibCompressor(
814 encoding=coding.value,
815 max_sync_chunk_size=self._zlib_executor_size,
816 executor=self._zlib_executor,
817 )
818 assert self._body is not None
819 if self._zlib_executor_size is None and len(self._body) > LARGE_BODY_SIZE:
820 warnings.warn(
821 "Synchronous compression of large response bodies "
822 f"({len(self._body)} bytes) might block the async event loop. "
823 "Consider providing a custom value to zlib_executor_size/"
824 "zlib_executor response properties or disabling compression on it."
825 )
826 self._compressed_body = (
827 await compressor.compress(self._body) + compressor.flush()
828 )
829 self._headers[hdrs.CONTENT_ENCODING] = coding.value
830 self._headers[hdrs.CONTENT_LENGTH] = str(len(self._compressed_body))
831
832
833def json_response(
834 data: Any = sentinel,
835 *,
836 text: Optional[str] = None,
837 body: Optional[bytes] = None,
838 status: int = 200,
839 reason: Optional[str] = None,
840 headers: Optional[LooseHeaders] = None,
841 content_type: str = "application/json",
842 dumps: JSONEncoder = json.dumps,
843) -> Response:
844 if data is not sentinel:
845 if text or body:
846 raise ValueError("only one of data, text, or body should be specified")
847 else:
848 text = dumps(data)
849 return Response(
850 text=text,
851 body=body,
852 status=status,
853 reason=reason,
854 headers=headers,
855 content_type=content_type,
856 )
857