Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/werkzeug/wrappers/response.py: 56%
274 statements
« prev ^ index » next coverage.py v7.3.2, created at 2023-12-09 07:17 +0000
1from __future__ import annotations
3import json
4import typing as t
5from http import HTTPStatus
6from urllib.parse import urljoin
8from ..datastructures import Headers
9from ..http import remove_entity_headers
10from ..sansio.response import Response as _SansIOResponse
11from ..urls import _invalid_iri_to_uri
12from ..urls import iri_to_uri
13from ..utils import cached_property
14from ..wsgi import ClosingIterator
15from ..wsgi import get_current_url
16from werkzeug._internal import _get_environ
17from werkzeug.http import generate_etag
18from werkzeug.http import http_date
19from werkzeug.http import is_resource_modified
20from werkzeug.http import parse_etags
21from werkzeug.http import parse_range_header
22from werkzeug.wsgi import _RangeWrapper
24if t.TYPE_CHECKING:
25 from _typeshed.wsgi import StartResponse
26 from _typeshed.wsgi import WSGIApplication
27 from _typeshed.wsgi import WSGIEnvironment
28 from .request import Request
31def _iter_encoded(iterable: t.Iterable[str | bytes]) -> t.Iterator[bytes]:
32 for item in iterable:
33 if isinstance(item, str):
34 yield item.encode()
35 else:
36 yield item
39class Response(_SansIOResponse):
40 """Represents an outgoing WSGI HTTP response with body, status, and
41 headers. Has properties and methods for using the functionality
42 defined by various HTTP specs.
44 The response body is flexible to support different use cases. The
45 simple form is passing bytes, or a string which will be encoded as
46 UTF-8. Passing an iterable of bytes or strings makes this a
47 streaming response. A generator is particularly useful for building
48 a CSV file in memory or using SSE (Server Sent Events). A file-like
49 object is also iterable, although the
50 :func:`~werkzeug.utils.send_file` helper should be used in that
51 case.
53 The response object is itself a WSGI application callable. When
54 called (:meth:`__call__`) with ``environ`` and ``start_response``,
55 it will pass its status and headers to ``start_response`` then
56 return its body as an iterable.
58 .. code-block:: python
60 from werkzeug.wrappers.response import Response
62 def index():
63 return Response("Hello, World!")
65 def application(environ, start_response):
66 path = environ.get("PATH_INFO") or "/"
68 if path == "/":
69 response = index()
70 else:
71 response = Response("Not Found", status=404)
73 return response(environ, start_response)
75 :param response: The data for the body of the response. A string or
76 bytes, or tuple or list of strings or bytes, for a fixed-length
77 response, or any other iterable of strings or bytes for a
78 streaming response. Defaults to an empty body.
79 :param status: The status code for the response. Either an int, in
80 which case the default status message is added, or a string in
81 the form ``{code} {message}``, like ``404 Not Found``. Defaults
82 to 200.
83 :param headers: A :class:`~werkzeug.datastructures.Headers` object,
84 or a list of ``(key, value)`` tuples that will be converted to a
85 ``Headers`` object.
86 :param mimetype: The mime type (content type without charset or
87 other parameters) of the response. If the value starts with
88 ``text/`` (or matches some other special cases), the charset
89 will be added to create the ``content_type``.
90 :param content_type: The full content type of the response.
91 Overrides building the value from ``mimetype``.
92 :param direct_passthrough: Pass the response body directly through
93 as the WSGI iterable. This can be used when the body is a binary
94 file or other iterator of bytes, to skip some unnecessary
95 checks. Use :func:`~werkzeug.utils.send_file` instead of setting
96 this manually.
98 .. versionchanged:: 2.1
99 Old ``BaseResponse`` and mixin classes were removed.
101 .. versionchanged:: 2.0
102 Combine ``BaseResponse`` and mixins into a single ``Response``
103 class.
105 .. versionchanged:: 0.5
106 The ``direct_passthrough`` parameter was added.
107 """
109 #: if set to `False` accessing properties on the response object will
110 #: not try to consume the response iterator and convert it into a list.
111 #:
112 #: .. versionadded:: 0.6.2
113 #:
114 #: That attribute was previously called `implicit_seqence_conversion`.
115 #: (Notice the typo). If you did use this feature, you have to adapt
116 #: your code to the name change.
117 implicit_sequence_conversion = True
119 #: If a redirect ``Location`` header is a relative URL, make it an
120 #: absolute URL, including scheme and domain.
121 #:
122 #: .. versionchanged:: 2.1
123 #: This is disabled by default, so responses will send relative
124 #: redirects.
125 #:
126 #: .. versionadded:: 0.8
127 autocorrect_location_header = False
129 #: Should this response object automatically set the content-length
130 #: header if possible? This is true by default.
131 #:
132 #: .. versionadded:: 0.8
133 automatically_set_content_length = True
135 #: The response body to send as the WSGI iterable. A list of strings
136 #: or bytes represents a fixed-length response, any other iterable
137 #: is a streaming response. Strings are encoded to bytes as UTF-8.
138 #:
139 #: Do not set to a plain string or bytes, that will cause sending
140 #: the response to be very inefficient as it will iterate one byte
141 #: at a time.
142 response: t.Iterable[str] | t.Iterable[bytes]
144 def __init__(
145 self,
146 response: t.Iterable[bytes] | bytes | t.Iterable[str] | str | None = None,
147 status: int | str | HTTPStatus | None = None,
148 headers: t.Mapping[str, str | t.Iterable[str]]
149 | t.Iterable[tuple[str, str]]
150 | None = None,
151 mimetype: str | None = None,
152 content_type: str | None = None,
153 direct_passthrough: bool = False,
154 ) -> None:
155 super().__init__(
156 status=status,
157 headers=headers,
158 mimetype=mimetype,
159 content_type=content_type,
160 )
162 #: Pass the response body directly through as the WSGI iterable.
163 #: This can be used when the body is a binary file or other
164 #: iterator of bytes, to skip some unnecessary checks. Use
165 #: :func:`~werkzeug.utils.send_file` instead of setting this
166 #: manually.
167 self.direct_passthrough = direct_passthrough
168 self._on_close: list[t.Callable[[], t.Any]] = []
170 # we set the response after the headers so that if a class changes
171 # the charset attribute, the data is set in the correct charset.
172 if response is None:
173 self.response = []
174 elif isinstance(response, (str, bytes, bytearray)):
175 self.set_data(response)
176 else:
177 self.response = response
179 def call_on_close(self, func: t.Callable[[], t.Any]) -> t.Callable[[], t.Any]:
180 """Adds a function to the internal list of functions that should
181 be called as part of closing down the response. Since 0.7 this
182 function also returns the function that was passed so that this
183 can be used as a decorator.
185 .. versionadded:: 0.6
186 """
187 self._on_close.append(func)
188 return func
190 def __repr__(self) -> str:
191 if self.is_sequence:
192 body_info = f"{sum(map(len, self.iter_encoded()))} bytes"
193 else:
194 body_info = "streamed" if self.is_streamed else "likely-streamed"
195 return f"<{type(self).__name__} {body_info} [{self.status}]>"
197 @classmethod
198 def force_type(
199 cls, response: Response, environ: WSGIEnvironment | None = None
200 ) -> Response:
201 """Enforce that the WSGI response is a response object of the current
202 type. Werkzeug will use the :class:`Response` internally in many
203 situations like the exceptions. If you call :meth:`get_response` on an
204 exception you will get back a regular :class:`Response` object, even
205 if you are using a custom subclass.
207 This method can enforce a given response type, and it will also
208 convert arbitrary WSGI callables into response objects if an environ
209 is provided::
211 # convert a Werkzeug response object into an instance of the
212 # MyResponseClass subclass.
213 response = MyResponseClass.force_type(response)
215 # convert any WSGI application into a response object
216 response = MyResponseClass.force_type(response, environ)
218 This is especially useful if you want to post-process responses in
219 the main dispatcher and use functionality provided by your subclass.
221 Keep in mind that this will modify response objects in place if
222 possible!
224 :param response: a response object or wsgi application.
225 :param environ: a WSGI environment object.
226 :return: a response object.
227 """
228 if not isinstance(response, Response):
229 if environ is None:
230 raise TypeError(
231 "cannot convert WSGI application into response"
232 " objects without an environ"
233 )
235 from ..test import run_wsgi_app
237 response = Response(*run_wsgi_app(response, environ))
239 response.__class__ = cls
240 return response
242 @classmethod
243 def from_app(
244 cls, app: WSGIApplication, environ: WSGIEnvironment, buffered: bool = False
245 ) -> Response:
246 """Create a new response object from an application output. This
247 works best if you pass it an application that returns a generator all
248 the time. Sometimes applications may use the `write()` callable
249 returned by the `start_response` function. This tries to resolve such
250 edge cases automatically. But if you don't get the expected output
251 you should set `buffered` to `True` which enforces buffering.
253 :param app: the WSGI application to execute.
254 :param environ: the WSGI environment to execute against.
255 :param buffered: set to `True` to enforce buffering.
256 :return: a response object.
257 """
258 from ..test import run_wsgi_app
260 return cls(*run_wsgi_app(app, environ, buffered))
262 @t.overload
263 def get_data(self, as_text: t.Literal[False] = False) -> bytes:
264 ...
266 @t.overload
267 def get_data(self, as_text: t.Literal[True]) -> str:
268 ...
270 def get_data(self, as_text: bool = False) -> bytes | str:
271 """The string representation of the response body. Whenever you call
272 this property the response iterable is encoded and flattened. This
273 can lead to unwanted behavior if you stream big data.
275 This behavior can be disabled by setting
276 :attr:`implicit_sequence_conversion` to `False`.
278 If `as_text` is set to `True` the return value will be a decoded
279 string.
281 .. versionadded:: 0.9
282 """
283 self._ensure_sequence()
284 rv = b"".join(self.iter_encoded())
286 if as_text:
287 return rv.decode()
289 return rv
291 def set_data(self, value: bytes | str) -> None:
292 """Sets a new string as response. The value must be a string or
293 bytes. If a string is set it's encoded to the charset of the
294 response (utf-8 by default).
296 .. versionadded:: 0.9
297 """
298 if isinstance(value, str):
299 value = value.encode()
300 self.response = [value]
301 if self.automatically_set_content_length:
302 self.headers["Content-Length"] = str(len(value))
304 data = property(
305 get_data,
306 set_data,
307 doc="A descriptor that calls :meth:`get_data` and :meth:`set_data`.",
308 )
310 def calculate_content_length(self) -> int | None:
311 """Returns the content length if available or `None` otherwise."""
312 try:
313 self._ensure_sequence()
314 except RuntimeError:
315 return None
316 return sum(len(x) for x in self.iter_encoded())
318 def _ensure_sequence(self, mutable: bool = False) -> None:
319 """This method can be called by methods that need a sequence. If
320 `mutable` is true, it will also ensure that the response sequence
321 is a standard Python list.
323 .. versionadded:: 0.6
324 """
325 if self.is_sequence:
326 # if we need a mutable object, we ensure it's a list.
327 if mutable and not isinstance(self.response, list):
328 self.response = list(self.response) # type: ignore
329 return
330 if self.direct_passthrough:
331 raise RuntimeError(
332 "Attempted implicit sequence conversion but the"
333 " response object is in direct passthrough mode."
334 )
335 if not self.implicit_sequence_conversion:
336 raise RuntimeError(
337 "The response object required the iterable to be a"
338 " sequence, but the implicit conversion was disabled."
339 " Call make_sequence() yourself."
340 )
341 self.make_sequence()
343 def make_sequence(self) -> None:
344 """Converts the response iterator in a list. By default this happens
345 automatically if required. If `implicit_sequence_conversion` is
346 disabled, this method is not automatically called and some properties
347 might raise exceptions. This also encodes all the items.
349 .. versionadded:: 0.6
350 """
351 if not self.is_sequence:
352 # if we consume an iterable we have to ensure that the close
353 # method of the iterable is called if available when we tear
354 # down the response
355 close = getattr(self.response, "close", None)
356 self.response = list(self.iter_encoded())
357 if close is not None:
358 self.call_on_close(close)
360 def iter_encoded(self) -> t.Iterator[bytes]:
361 """Iter the response encoded with the encoding of the response.
362 If the response object is invoked as WSGI application the return
363 value of this method is used as application iterator unless
364 :attr:`direct_passthrough` was activated.
365 """
366 # Encode in a separate function so that self.response is fetched
367 # early. This allows us to wrap the response with the return
368 # value from get_app_iter or iter_encoded.
369 return _iter_encoded(self.response)
371 @property
372 def is_streamed(self) -> bool:
373 """If the response is streamed (the response is not an iterable with
374 a length information) this property is `True`. In this case streamed
375 means that there is no information about the number of iterations.
376 This is usually `True` if a generator is passed to the response object.
378 This is useful for checking before applying some sort of post
379 filtering that should not take place for streamed responses.
380 """
381 try:
382 len(self.response) # type: ignore
383 except (TypeError, AttributeError):
384 return True
385 return False
387 @property
388 def is_sequence(self) -> bool:
389 """If the iterator is buffered, this property will be `True`. A
390 response object will consider an iterator to be buffered if the
391 response attribute is a list or tuple.
393 .. versionadded:: 0.6
394 """
395 return isinstance(self.response, (tuple, list))
397 def close(self) -> None:
398 """Close the wrapped response if possible. You can also use the object
399 in a with statement which will automatically close it.
401 .. versionadded:: 0.9
402 Can now be used in a with statement.
403 """
404 if hasattr(self.response, "close"):
405 self.response.close()
406 for func in self._on_close:
407 func()
409 def __enter__(self) -> Response:
410 return self
412 def __exit__(self, exc_type, exc_value, tb): # type: ignore
413 self.close()
415 def freeze(self) -> None:
416 """Make the response object ready to be pickled. Does the
417 following:
419 * Buffer the response into a list, ignoring
420 :attr:`implicity_sequence_conversion` and
421 :attr:`direct_passthrough`.
422 * Set the ``Content-Length`` header.
423 * Generate an ``ETag`` header if one is not already set.
425 .. versionchanged:: 2.1
426 Removed the ``no_etag`` parameter.
428 .. versionchanged:: 2.0
429 An ``ETag`` header is always added.
431 .. versionchanged:: 0.6
432 The ``Content-Length`` header is set.
433 """
434 # Always freeze the encoded response body, ignore
435 # implicit_sequence_conversion and direct_passthrough.
436 self.response = list(self.iter_encoded())
437 self.headers["Content-Length"] = str(sum(map(len, self.response)))
438 self.add_etag()
440 def get_wsgi_headers(self, environ: WSGIEnvironment) -> Headers:
441 """This is automatically called right before the response is started
442 and returns headers modified for the given environment. It returns a
443 copy of the headers from the response with some modifications applied
444 if necessary.
446 For example the location header (if present) is joined with the root
447 URL of the environment. Also the content length is automatically set
448 to zero here for certain status codes.
450 .. versionchanged:: 0.6
451 Previously that function was called `fix_headers` and modified
452 the response object in place. Also since 0.6, IRIs in location
453 and content-location headers are handled properly.
455 Also starting with 0.6, Werkzeug will attempt to set the content
456 length if it is able to figure it out on its own. This is the
457 case if all the strings in the response iterable are already
458 encoded and the iterable is buffered.
460 :param environ: the WSGI environment of the request.
461 :return: returns a new :class:`~werkzeug.datastructures.Headers`
462 object.
463 """
464 headers = Headers(self.headers)
465 location: str | None = None
466 content_location: str | None = None
467 content_length: str | int | None = None
468 status = self.status_code
470 # iterate over the headers to find all values in one go. Because
471 # get_wsgi_headers is used each response that gives us a tiny
472 # speedup.
473 for key, value in headers:
474 ikey = key.lower()
475 if ikey == "location":
476 location = value
477 elif ikey == "content-location":
478 content_location = value
479 elif ikey == "content-length":
480 content_length = value
482 if location is not None:
483 location = _invalid_iri_to_uri(location)
485 if self.autocorrect_location_header:
486 # Make the location header an absolute URL.
487 current_url = get_current_url(environ, strip_querystring=True)
488 current_url = iri_to_uri(current_url)
489 location = urljoin(current_url, location)
491 headers["Location"] = location
493 # make sure the content location is a URL
494 if content_location is not None:
495 headers["Content-Location"] = iri_to_uri(content_location)
497 if 100 <= status < 200 or status == 204:
498 # Per section 3.3.2 of RFC 7230, "a server MUST NOT send a
499 # Content-Length header field in any response with a status
500 # code of 1xx (Informational) or 204 (No Content)."
501 headers.remove("Content-Length")
502 elif status == 304:
503 remove_entity_headers(headers)
505 # if we can determine the content length automatically, we
506 # should try to do that. But only if this does not involve
507 # flattening the iterator or encoding of strings in the
508 # response. We however should not do that if we have a 304
509 # response.
510 if (
511 self.automatically_set_content_length
512 and self.is_sequence
513 and content_length is None
514 and status not in (204, 304)
515 and not (100 <= status < 200)
516 ):
517 content_length = sum(len(x) for x in self.iter_encoded())
518 headers["Content-Length"] = str(content_length)
520 return headers
522 def get_app_iter(self, environ: WSGIEnvironment) -> t.Iterable[bytes]:
523 """Returns the application iterator for the given environ. Depending
524 on the request method and the current status code the return value
525 might be an empty response rather than the one from the response.
527 If the request method is `HEAD` or the status code is in a range
528 where the HTTP specification requires an empty response, an empty
529 iterable is returned.
531 .. versionadded:: 0.6
533 :param environ: the WSGI environment of the request.
534 :return: a response iterable.
535 """
536 status = self.status_code
537 if (
538 environ["REQUEST_METHOD"] == "HEAD"
539 or 100 <= status < 200
540 or status in (204, 304)
541 ):
542 iterable: t.Iterable[bytes] = ()
543 elif self.direct_passthrough:
544 return self.response # type: ignore
545 else:
546 iterable = self.iter_encoded()
547 return ClosingIterator(iterable, self.close)
549 def get_wsgi_response(
550 self, environ: WSGIEnvironment
551 ) -> tuple[t.Iterable[bytes], str, list[tuple[str, str]]]:
552 """Returns the final WSGI response as tuple. The first item in
553 the tuple is the application iterator, the second the status and
554 the third the list of headers. The response returned is created
555 specially for the given environment. For example if the request
556 method in the WSGI environment is ``'HEAD'`` the response will
557 be empty and only the headers and status code will be present.
559 .. versionadded:: 0.6
561 :param environ: the WSGI environment of the request.
562 :return: an ``(app_iter, status, headers)`` tuple.
563 """
564 headers = self.get_wsgi_headers(environ)
565 app_iter = self.get_app_iter(environ)
566 return app_iter, self.status, headers.to_wsgi_list()
568 def __call__(
569 self, environ: WSGIEnvironment, start_response: StartResponse
570 ) -> t.Iterable[bytes]:
571 """Process this response as WSGI application.
573 :param environ: the WSGI environment.
574 :param start_response: the response callable provided by the WSGI
575 server.
576 :return: an application iterator
577 """
578 app_iter, status, headers = self.get_wsgi_response(environ)
579 start_response(status, headers)
580 return app_iter
582 # JSON
584 #: A module or other object that has ``dumps`` and ``loads``
585 #: functions that match the API of the built-in :mod:`json` module.
586 json_module = json
588 @property
589 def json(self) -> t.Any | None:
590 """The parsed JSON data if :attr:`mimetype` indicates JSON
591 (:mimetype:`application/json`, see :attr:`is_json`).
593 Calls :meth:`get_json` with default arguments.
594 """
595 return self.get_json()
597 @t.overload
598 def get_json(self, force: bool = ..., silent: t.Literal[False] = ...) -> t.Any:
599 ...
601 @t.overload
602 def get_json(self, force: bool = ..., silent: bool = ...) -> t.Any | None:
603 ...
605 def get_json(self, force: bool = False, silent: bool = False) -> t.Any | None:
606 """Parse :attr:`data` as JSON. Useful during testing.
608 If the mimetype does not indicate JSON
609 (:mimetype:`application/json`, see :attr:`is_json`), this
610 returns ``None``.
612 Unlike :meth:`Request.get_json`, the result is not cached.
614 :param force: Ignore the mimetype and always try to parse JSON.
615 :param silent: Silence parsing errors and return ``None``
616 instead.
617 """
618 if not (force or self.is_json):
619 return None
621 data = self.get_data()
623 try:
624 return self.json_module.loads(data)
625 except ValueError:
626 if not silent:
627 raise
629 return None
631 # Stream
633 @cached_property
634 def stream(self) -> ResponseStream:
635 """The response iterable as write-only stream."""
636 return ResponseStream(self)
638 def _wrap_range_response(self, start: int, length: int) -> None:
639 """Wrap existing Response in case of Range Request context."""
640 if self.status_code == 206:
641 self.response = _RangeWrapper(self.response, start, length) # type: ignore
643 def _is_range_request_processable(self, environ: WSGIEnvironment) -> bool:
644 """Return ``True`` if `Range` header is present and if underlying
645 resource is considered unchanged when compared with `If-Range` header.
646 """
647 return (
648 "HTTP_IF_RANGE" not in environ
649 or not is_resource_modified(
650 environ,
651 self.headers.get("etag"),
652 None,
653 self.headers.get("last-modified"),
654 ignore_if_range=False,
655 )
656 ) and "HTTP_RANGE" in environ
658 def _process_range_request(
659 self,
660 environ: WSGIEnvironment,
661 complete_length: int | None,
662 accept_ranges: bool | str,
663 ) -> bool:
664 """Handle Range Request related headers (RFC7233). If `Accept-Ranges`
665 header is valid, and Range Request is processable, we set the headers
666 as described by the RFC, and wrap the underlying response in a
667 RangeWrapper.
669 Returns ``True`` if Range Request can be fulfilled, ``False`` otherwise.
671 :raises: :class:`~werkzeug.exceptions.RequestedRangeNotSatisfiable`
672 if `Range` header could not be parsed or satisfied.
674 .. versionchanged:: 2.0
675 Returns ``False`` if the length is 0.
676 """
677 from ..exceptions import RequestedRangeNotSatisfiable
679 if (
680 not accept_ranges
681 or complete_length is None
682 or complete_length == 0
683 or not self._is_range_request_processable(environ)
684 ):
685 return False
687 if accept_ranges is True:
688 accept_ranges = "bytes"
690 parsed_range = parse_range_header(environ.get("HTTP_RANGE"))
692 if parsed_range is None:
693 raise RequestedRangeNotSatisfiable(complete_length)
695 range_tuple = parsed_range.range_for_length(complete_length)
696 content_range_header = parsed_range.to_content_range_header(complete_length)
698 if range_tuple is None or content_range_header is None:
699 raise RequestedRangeNotSatisfiable(complete_length)
701 content_length = range_tuple[1] - range_tuple[0]
702 self.headers["Content-Length"] = str(content_length)
703 self.headers["Accept-Ranges"] = accept_ranges
704 self.content_range = content_range_header # type: ignore
705 self.status_code = 206
706 self._wrap_range_response(range_tuple[0], content_length)
707 return True
709 def make_conditional(
710 self,
711 request_or_environ: WSGIEnvironment | Request,
712 accept_ranges: bool | str = False,
713 complete_length: int | None = None,
714 ) -> Response:
715 """Make the response conditional to the request. This method works
716 best if an etag was defined for the response already. The `add_etag`
717 method can be used to do that. If called without etag just the date
718 header is set.
720 This does nothing if the request method in the request or environ is
721 anything but GET or HEAD.
723 For optimal performance when handling range requests, it's recommended
724 that your response data object implements `seekable`, `seek` and `tell`
725 methods as described by :py:class:`io.IOBase`. Objects returned by
726 :meth:`~werkzeug.wsgi.wrap_file` automatically implement those methods.
728 It does not remove the body of the response because that's something
729 the :meth:`__call__` function does for us automatically.
731 Returns self so that you can do ``return resp.make_conditional(req)``
732 but modifies the object in-place.
734 :param request_or_environ: a request object or WSGI environment to be
735 used to make the response conditional
736 against.
737 :param accept_ranges: This parameter dictates the value of
738 `Accept-Ranges` header. If ``False`` (default),
739 the header is not set. If ``True``, it will be set
740 to ``"bytes"``. If it's a string, it will use this
741 value.
742 :param complete_length: Will be used only in valid Range Requests.
743 It will set `Content-Range` complete length
744 value and compute `Content-Length` real value.
745 This parameter is mandatory for successful
746 Range Requests completion.
747 :raises: :class:`~werkzeug.exceptions.RequestedRangeNotSatisfiable`
748 if `Range` header could not be parsed or satisfied.
750 .. versionchanged:: 2.0
751 Range processing is skipped if length is 0 instead of
752 raising a 416 Range Not Satisfiable error.
753 """
754 environ = _get_environ(request_or_environ)
755 if environ["REQUEST_METHOD"] in ("GET", "HEAD"):
756 # if the date is not in the headers, add it now. We however
757 # will not override an already existing header. Unfortunately
758 # this header will be overridden by many WSGI servers including
759 # wsgiref.
760 if "date" not in self.headers:
761 self.headers["Date"] = http_date()
762 is206 = self._process_range_request(environ, complete_length, accept_ranges)
763 if not is206 and not is_resource_modified(
764 environ,
765 self.headers.get("etag"),
766 None,
767 self.headers.get("last-modified"),
768 ):
769 if parse_etags(environ.get("HTTP_IF_MATCH")):
770 self.status_code = 412
771 else:
772 self.status_code = 304
773 if (
774 self.automatically_set_content_length
775 and "content-length" not in self.headers
776 ):
777 length = self.calculate_content_length()
778 if length is not None:
779 self.headers["Content-Length"] = str(length)
780 return self
782 def add_etag(self, overwrite: bool = False, weak: bool = False) -> None:
783 """Add an etag for the current response if there is none yet.
785 .. versionchanged:: 2.0
786 SHA-1 is used to generate the value. MD5 may not be
787 available in some environments.
788 """
789 if overwrite or "etag" not in self.headers:
790 self.set_etag(generate_etag(self.get_data()), weak)
793class ResponseStream:
794 """A file descriptor like object used by :meth:`Response.stream` to
795 represent the body of the stream. It directly pushes into the
796 response iterable of the response object.
797 """
799 mode = "wb+"
801 def __init__(self, response: Response):
802 self.response = response
803 self.closed = False
805 def write(self, value: bytes) -> int:
806 if self.closed:
807 raise ValueError("I/O operation on closed file")
808 self.response._ensure_sequence(mutable=True)
809 self.response.response.append(value) # type: ignore
810 self.response.headers.pop("Content-Length", None)
811 return len(value)
813 def writelines(self, seq: t.Iterable[bytes]) -> None:
814 for item in seq:
815 self.write(item)
817 def close(self) -> None:
818 self.closed = True
820 def flush(self) -> None:
821 if self.closed:
822 raise ValueError("I/O operation on closed file")
824 def isatty(self) -> bool:
825 if self.closed:
826 raise ValueError("I/O operation on closed file")
827 return False
829 def tell(self) -> int:
830 self.response._ensure_sequence()
831 return sum(map(len, self.response.response))
833 @property
834 def encoding(self) -> str:
835 return "utf-8"