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