webob.response のソースコード

import re
import struct
import zlib
from base64 import b64encode
from datetime import datetime, timedelta
from hashlib import md5

from webob.byterange import ContentRange
from webob.cachecontrol import CacheControl, serialize_cache_control
from webob.compat import (
    PY2,
    bytes_,
    native_,
    string_types,
    text_type,
    url_quote,
    urlparse,
)
from webob.cookies import Cookie, make_cookie
from webob.datetime_utils import (
    parse_date_delta,
    serialize_date_delta,
    timedelta_to_seconds,
)
from webob.descriptors import (
    CHARSET_RE,
    SCHEME_RE,
    converter,
    date_header,
    header_getter,
    list_header,
    parse_auth,
    parse_content_range,
    parse_etag_response,
    parse_int,
    parse_int_safe,
    serialize_auth,
    serialize_content_range,
    serialize_etag_response,
    serialize_int,
)
from webob.headers import ResponseHeaders
from webob.request import BaseRequest
from webob.util import status_generic_reasons, status_reasons, warn_deprecation

try:
    import simplejson as json
except ImportError:
    import json

__all__ = ['Response']

_PARAM_RE = re.compile(r'([a-z0-9]+)=(?:"([^"]*)"|([a-z0-9_.-]*))', re.I)
_OK_PARAM_RE = re.compile(r'^[a-z0-9_.-]+$', re.I)

_gzip_header = b'\x1f\x8b\x08\x00\x00\x00\x00\x00\x02\xff'

_marker = object()

class Response(object):
    """
    Represents a WSGI response.

    If no arguments are passed, creates a :class:`~Response` that uses a
    variety of defaults. The defaults may be changed by sub-classing the
    :class:`~Response`. See the :ref:`sub-classing notes
    <response_subclassing_notes>`.

    :cvar ~Response.body: If ``body`` is a ``text_type``, then it will be
        encoded using either ``charset`` when provided or ``default_encoding``
        when ``charset`` is not provided if the ``content_type`` allows for a
        ``charset``. This argument is mutually  exclusive with ``app_iter``.

    :vartype ~Response.body: bytes or text_type

    :cvar ~Response.status: Either an :class:`int` or a string that is
        an integer followed by the status text. If it is an integer, it will be
        converted to a proper status that also includes the status text.  Any
        existing status text will be kept. Non-standard values are allowed.

    :vartype ~Response.status: int or str

    :cvar ~Response.headerlist: A list of HTTP headers for the response.

    :vartype ~Response.headerlist: list

    :cvar ~Response.app_iter: An iterator that is used as the body of the
        response. Should conform to the WSGI requirements and should provide
        bytes. This argument is mutually exclusive with ``body``.

    :vartype ~Response.app_iter: iterable

    :cvar ~Response.content_type: Sets the ``Content-Type`` header. If no
        ``content_type`` is provided, and there is no ``headerlist``, the
        ``default_content_type`` will be automatically set. If ``headerlist``
        is provided then this value is ignored.

    :vartype ~Response.content_type: str or None

    :cvar conditional_response: Used to change the behavior of the
        :class:`~Response` to check the original request for conditional
        response headers. See :meth:`~Response.conditional_response_app` for
        more information.

    :vartype conditional_response: bool

    :cvar ~Response.charset: Adds a ``charset`` ``Content-Type`` parameter. If
        no ``charset`` is provided and the ``Content-Type`` is text, then the
        ``default_charset`` will automatically be added.  Currently the only
        ``Content-Type``'s that allow for a ``charset`` are defined to be
        ``text/*``, ``application/xml``, and ``*/*+xml``. Any other
        ``Content-Type``'s will not have a ``charset`` added. If a
        ``headerlist`` is provided this value is ignored.

    :vartype ~Response.charset: str or None

    All other response attributes may be set on the response by providing them
    as keyword arguments. A :exc:`TypeError` will be raised for any unexpected
    keywords.

    .. _response_subclassing_notes:

    **Sub-classing notes:**

    * The ``default_content_type`` is used as the default for the
      ``Content-Type`` header that is returned on the response. It is
      ``text/html``.

    * The ``default_charset`` is used as the default character set to return on
      the ``Content-Type`` header, if the ``Content-Type`` allows for a
      ``charset`` parameter. Currently the only ``Content-Type``'s that allow
      for a ``charset`` are defined to be: ``text/*``, ``application/xml``, and
      ``*/*+xml``. Any other ``Content-Type``'s will not have a ``charset``
      added.

    * The ``unicode_errors`` is set to ``strict``, and access on a
      :attr:`~Response.text` will raise an error if it fails to decode the
      :attr:`~Response.body`.

    * ``default_conditional_response`` is set to ``False``. This flag may be
      set to ``True`` so that all ``Response`` objects will attempt to check
      the original request for conditional response headers. See
      :meth:`~Response.conditional_response_app` for more information.

    * ``default_body_encoding`` is set to 'UTF-8' by default. It exists to
      allow users to get/set the ``Response`` object using ``.text``, even if
      no ``charset`` has been set for the ``Content-Type``.
    """

    default_content_type = 'text/html'
    default_charset = 'UTF-8'
    unicode_errors = 'strict'
    default_conditional_response = False
    default_body_encoding = 'UTF-8'

    # These two are only around so that when people pass them into the
    # constructor they correctly get saved and set, however they are not used
    # by any part of the Response. See commit
    # 627593bbcd4ab52adc7ee569001cdda91c670d5d for rationale.
    request = None
    environ = None

    #
    # __init__, from_file, copy
    #

    def __init__(self, body=None, status=None, headerlist=None, app_iter=None,
                 content_type=None, conditional_response=None, charset=_marker,
                 **kw):
        # Do some sanity checking, and turn json_body into an actual body
        if app_iter is None and body is None and ('json_body' in kw or 'json' in kw):
            if 'json_body' in kw:
                json_body = kw.pop('json_body')
            else:
                json_body = kw.pop('json')
            body = json.dumps(json_body, separators=(',', ':')).encode('UTF-8')

            if content_type is None:
                content_type = 'application/json'

        if app_iter is None:
            if body is None:
                body = b''
        elif body is not None:
            raise TypeError(
                "You may only give one of the body and app_iter arguments")

        # Set up Response.status
        if status is None:
            self._status = '200 OK'
        else:
            self.status = status

        # Initialize headers
        self._headers = None
        if headerlist is None:
            self._headerlist = []
        else:
            self._headerlist = headerlist

        # Set the encoding for the Response to charset, so if a charset is
        # passed but the Content-Type does not allow for a charset, we can
        # still encode text_type body's.
        # r = Response(
        #   content_type='application/foo',
        #   charset='UTF-8',
        #   body=u'somebody')
        # Should work without issues, and the header will be correctly set to
        # Content-Type: application/foo with no charset on it.

        encoding = None
        if charset is not _marker:
            encoding = charset

        # Does the status code have a body or not?
        code_has_body = (
            self._status[0] != '1' and
            self._status[:3] not in ('204', '205', '304')
        )

        # We only set the content_type to the one passed to the constructor or
        # the default content type if there is none that exists AND there was
        # no headerlist passed. If a headerlist was provided then most likely
        # the ommission of the Content-Type is on purpose and we shouldn't try
        # to be smart about it.
        #
        # Also allow creation of a empty Response with just the status set to a
        # Response with empty body, such as Response(status='204 No Content')
        # without the default content_type being set (since empty bodies have
        # no Content-Type)
        #
        # Check if content_type is set because default_content_type could be
        # None, in which case there is no content_type, and thus we don't need
        # to anything

        content_type = content_type or self.default_content_type

        if headerlist is None and code_has_body and content_type:
            # Set up the charset, if the content_type doesn't already have one

            has_charset = 'charset=' in content_type

            # If the Content-Type already has a charset, we don't set the user
            # provided charset on the Content-Type, so we shouldn't use it as
            # the encoding for text_type based body's.
            if has_charset:
                encoding = None

            # Do not use the default_charset for the encoding because we
            # want things like
            # Response(content_type='image/jpeg',body=u'foo') to raise when
            # trying to encode the body.

            new_charset = encoding

            if (
                not has_charset and
                charset is _marker and
                self.default_charset
            ):
                new_charset = self.default_charset

            # Optimize for the default_content_type as shipped by
            # WebOb, becuase we know that 'text/html' has a charset,
            # otherwise add a charset if the content_type has a charset.
            #
            # Even if the user supplied charset explicitly, we do not add
            # it to the Content-Type unless it has has a charset, instead
            # the user supplied charset is solely used for encoding the
            # body if it is a text_type

            if (
                new_charset and
                (
                    content_type == 'text/html' or
                    _content_type_has_charset(content_type)
                )
            ):
                content_type += '; charset=' + new_charset

            self._headerlist.append(('Content-Type', content_type))

        # Set up conditional response
        if conditional_response is None:
            self.conditional_response = self.default_conditional_response
        else:
            self.conditional_response = bool(conditional_response)

        # Set up app_iter if the HTTP Status code has a body
        if app_iter is None and code_has_body:
            if isinstance(body, text_type):
                # Fall back to trying self.charset if encoding is not set. In
                # most cases encoding will be set to the default value.
                encoding = encoding or self.charset
                if encoding is None:
                    raise TypeError(
                        "You cannot set the body to a text value without a "
                        "charset")
                body = body.encode(encoding)
            app_iter = [body]

            if headerlist is not None:
                self._headerlist[:] = [
                    (k, v)
                    for (k, v)
                    in self._headerlist
                    if k.lower() != 'content-length'
                ]
            self._headerlist.append(('Content-Length', str(len(body))))
        elif app_iter is None and not code_has_body:
            app_iter = [b'']

        self._app_iter = app_iter

        # Loop through all the remaining keyword arguments
        for name, value in kw.items():
            if not hasattr(self.__class__, name):
                # Not a basic attribute
                raise TypeError(
                    "Unexpected keyword: %s=%r" % (name, value))
            setattr(self, name, value)

    @classmethod
    def from_file(cls, fp):
        """Reads a response from a file-like object (it must implement
        ``.read(size)`` and ``.readline()``).

        It will read up to the end of the response, not the end of the
        file.

        This reads the response as represented by ``str(resp)``; it
        may not read every valid HTTP response properly.  Responses
        must have a ``Content-Length``."""
        headerlist = []
        status = fp.readline().strip()
        is_text = isinstance(status, text_type)

        if is_text:
            _colon = ':'
            _http = 'HTTP/'
        else:
            _colon = b':'
            _http = b'HTTP/'

        if status.startswith(_http):
            (http_ver, status_num, status_text) = status.split(None, 2)
            status = '%s %s' % (native_(status_num), native_(status_text))

        while 1:
            line = fp.readline().strip()
            if not line:
                # end of headers
                break
            try:
                header_name, value = line.split(_colon, 1)
            except ValueError:
                raise ValueError('Bad header line: %r' % line)
            value = value.strip()
            headerlist.append((
                native_(header_name, 'latin-1'),
                native_(value, 'latin-1')
            ))
        r = cls(
            status=status,
            headerlist=headerlist,
            app_iter=(),
        )
        body = fp.read(r.content_length or 0)
        if is_text:
            r.text = body
        else:
            r.body = body
        return r

    def copy(self):
        """Makes a copy of the response."""
        # we need to do this for app_iter to be reusable
        app_iter = list(self._app_iter)
        iter_close(self._app_iter)
        # and this to make sure app_iter instances are different
        self._app_iter = list(app_iter)
        return self.__class__(
            status=self._status,
            headerlist=self._headerlist[:],
            app_iter=app_iter,
            conditional_response=self.conditional_response)

    #
    # __repr__, __str__
    #

    def __repr__(self):
        return '<%s at 0x%x %s>' % (self.__class__.__name__, abs(id(self)),
                                    self.status)

    def __str__(self, skip_body=False):
        parts = [self.status]
        if not skip_body:
            # Force enumeration of the body (to set content-length)
            self.body
        parts += map('%s: %s'.__mod__, self.headerlist)
        if not skip_body and self.body:
            parts += ['', self.body if PY2 else self.text]
        return '\r\n'.join(parts)

    #
    # status, status_code/status_int
    #

    def _status__get(self):
        """
        The status string.
        """
        return self._status

    def _status__set(self, value):
        try:
            code = int(value)
        except (ValueError, TypeError):
            pass
        else:
            self.status_code = code
            return
        if not PY2:
            if isinstance(value, bytes):
                value = value.decode('ascii')
        elif isinstance(value, text_type):
            value = value.encode('ascii')
        if not isinstance(value, str):
            raise TypeError(
                "You must set status to a string or integer (not %s)"
                % type(value))

        # Attempt to get the status code itself, if this fails we should fail
        try:
            # We don't need this value anywhere, we just want to validate it's
            # an integer. So we are using the side-effect of int() raises a
            # ValueError as a test
            int(value.split()[0])
        except ValueError:
            raise ValueError('Invalid status code, integer required.')
        self._status = value

    status = property(_status__get, _status__set, doc=_status__get.__doc__)

    def _status_code__get(self):
        """
        The status as an integer.
        """
        return int(self._status.split()[0])

    def _status_code__set(self, code):
        try:
            self._status = '%d %s' % (code, status_reasons[code])
        except KeyError:
            self._status = '%d %s' % (code, status_generic_reasons[code // 100])

    status_code = status_int = property(_status_code__get, _status_code__set,
                                        doc=_status_code__get.__doc__)

    #
    # headerslist, headers
    #

    def _headerlist__get(self):
        """
        The list of response headers.
        """
        return self._headerlist

    def _headerlist__set(self, value):
        self._headers = None
        if not isinstance(value, list):
            if hasattr(value, 'items'):
                value = value.items()
            value = list(value)
        self._headerlist = value

    def _headerlist__del(self):
        self.headerlist = []

    headerlist = property(_headerlist__get, _headerlist__set,
                          _headerlist__del, doc=_headerlist__get.__doc__)

    def _headers__get(self):
        """
        The headers in a dictionary-like object.
        """
        if self._headers is None:
            self._headers = ResponseHeaders.view_list(self._headerlist)
        return self._headers

    def _headers__set(self, value):
        if hasattr(value, 'items'):
            value = value.items()
        self.headerlist = value
        self._headers = None

    headers = property(_headers__get, _headers__set, doc=_headers__get.__doc__)

    #
    # body
    #

    def _body__get(self):
        """
        The body of the response, as a :class:`bytes`.  This will read in
        the entire app_iter if necessary.
        """
        app_iter = self._app_iter
#         try:
#             if len(app_iter) == 1:
#                 return app_iter[0]
#         except:
#             pass
        if isinstance(app_iter, list) and len(app_iter) == 1:
            return app_iter[0]
        if app_iter is None:
            raise AttributeError("No body has been set")
        try:
            body = b''.join(app_iter)
        finally:
            iter_close(app_iter)
        if isinstance(body, text_type):
            raise _error_unicode_in_app_iter(app_iter, body)
        self._app_iter = [body]
        if len(body) == 0:
            # if body-length is zero, we assume it's a HEAD response and
            # leave content_length alone
            pass
        elif self.content_length is None:
            self.content_length = len(body)
        elif self.content_length != len(body):
            raise AssertionError(
                "Content-Length is different from actual app_iter length "
                "(%r!=%r)"
                % (self.content_length, len(body))
            )
        return body

    def _body__set(self, value=b''):
        if not isinstance(value, bytes):
            if isinstance(value, text_type):
                msg = ("You cannot set Response.body to a text object "
                       "(use Response.text)")
            else:
                msg = ("You can only set the body to a binary type (not %s)" %
                       type(value))
            raise TypeError(msg)
        if self._app_iter is not None:
            self.content_md5 = None
        self._app_iter = [value]
        self.content_length = len(value)

#     def _body__del(self):
#         self.body = ''
#         #self.content_length = None

    body = property(_body__get, _body__set, _body__set)

    def _json_body__get(self):
        """
        Set/get the body of the response as JSON.

        .. note::

           This will automatically :meth:`~bytes.decode` the
           :attr:`~Response.body` as ``UTF-8`` on get, and
           :meth:`~str.encode` the :meth:`json.dumps` as ``UTF-8``
           before assigning to :attr:`~Response.body`.

        """
        # Note: UTF-8 is a content-type specific default for JSON
        return json.loads(self.body.decode('UTF-8'))

    def _json_body__set(self, value):
        self.body = json.dumps(value, separators=(',', ':')).encode('UTF-8')

    def _json_body__del(self):
        del self.body

    json = json_body = property(_json_body__get, _json_body__set, _json_body__del)

    def _has_body__get(self):
        """
        Determine if the the response has a :attr:`~Response.body`. In
        contrast to simply accessing :attr:`~Response.body`, this method
        will **not** read the underlying :attr:`~Response.app_iter`.
        """

        app_iter = self._app_iter

        if isinstance(app_iter, list) and len(app_iter) == 1:
            if app_iter[0] != b'':
                return True
            else:
                return False

        if app_iter is None: # pragma: no cover
            return False

        return True

    has_body = property(_has_body__get)

    #
    # text, unicode_body, ubody
    #

    def _text__get(self):
        """
        Get/set the text value of the body using the ``charset`` of the
        ``Content-Type`` or the ``default_body_encoding``.
        """
        if not self.charset and not self.default_body_encoding:
            raise AttributeError(
                "You cannot access Response.text unless charset or default_body_encoding"
                " is set"
            )
        decoding = self.charset or self.default_body_encoding
        body = self.body
        return body.decode(decoding, self.unicode_errors)

    def _text__set(self, value):
        if not self.charset and not self.default_body_encoding:
            raise AttributeError(
                "You cannot access Response.text unless charset or default_body_encoding"
                " is set"
            )
        if not isinstance(value, text_type):
            raise TypeError(
                "You can only set Response.text to a unicode string "
                "(not %s)" % type(value))
        encoding = self.charset or self.default_body_encoding
        self.body = value.encode(encoding)

    def _text__del(self):
        del self.body

    text = property(_text__get, _text__set, _text__del, doc=_text__get.__doc__)

    unicode_body = ubody = property(_text__get, _text__set, _text__del,
                                    "Deprecated alias for .text")

    #
    # body_file, write(text)
    #

    def _body_file__get(self):
        """
        A file-like object that can be used to write to the
        body.  If you passed in a list ``app_iter``, that ``app_iter`` will be
        modified by writes.
        """
        return ResponseBodyFile(self)

    def _body_file__set(self, file):
        self.app_iter = iter_file(file)

    def _body_file__del(self):
        del self.body

    body_file = property(_body_file__get, _body_file__set, _body_file__del,
                         doc=_body_file__get.__doc__)

    def write(self, text):
        if not isinstance(text, bytes):
            if not isinstance(text, text_type):
                msg = "You can only write str to a Response.body_file, not %s"
                raise TypeError(msg % type(text))
            if not self.charset:
                msg = ("You can only write text to Response if charset has "
                       "been set")
                raise TypeError(msg)
            text = text.encode(self.charset)
        app_iter = self._app_iter
        if not isinstance(app_iter, list):
            try:
                new_app_iter = self._app_iter = list(app_iter)
            finally:
                iter_close(app_iter)
            app_iter = new_app_iter
            self.content_length = sum(len(chunk) for chunk in app_iter)
        app_iter.append(text)
        if self.content_length is not None:
            self.content_length += len(text)

    #
    # app_iter
    #

    def _app_iter__get(self):
        """
        Returns the ``app_iter`` of the response.

        If ``body`` was set, this will create an ``app_iter`` from that
        ``body`` (a single-item list).
        """
        return self._app_iter

    def _app_iter__set(self, value):
        if self._app_iter is not None:
            # Undo the automatically-set content-length
            self.content_length = None
        self._app_iter = value

    def _app_iter__del(self):
        self._app_iter = []
        self.content_length = None

    app_iter = property(_app_iter__get, _app_iter__set, _app_iter__del,
                        doc=_app_iter__get.__doc__)

    #
    # headers attrs
    #

    allow = list_header('Allow', '14.7')
    # TODO: (maybe) support response.vary += 'something'
    # TODO: same thing for all listy headers
    vary = list_header('Vary', '14.44')

    content_length = converter(
        header_getter('Content-Length', '14.17'),
        parse_int, serialize_int, 'int')

    content_encoding = header_getter('Content-Encoding', '14.11')
    content_language = list_header('Content-Language', '14.12')
    content_location = header_getter('Content-Location', '14.14')
    content_md5 = header_getter('Content-MD5', '14.14')
    content_disposition = header_getter('Content-Disposition', '19.5.1')

    accept_ranges = header_getter('Accept-Ranges', '14.5')
    content_range = converter(
        header_getter('Content-Range', '14.16'),
        parse_content_range, serialize_content_range, 'ContentRange object')

    date = date_header('Date', '14.18')
    expires = date_header('Expires', '14.21')
    last_modified = date_header('Last-Modified', '14.29')

    _etag_raw = header_getter('ETag', '14.19')
    etag = converter(
        _etag_raw,
        parse_etag_response, serialize_etag_response,
        'Entity tag'
    )
    @property
    def etag_strong(self):
        return parse_etag_response(self._etag_raw, strong=True)

    location = header_getter('Location', '14.30')
    pragma = header_getter('Pragma', '14.32')
    age = converter(
        header_getter('Age', '14.6'),
        parse_int_safe, serialize_int, 'int')

    retry_after = converter(
        header_getter('Retry-After', '14.37'),
        parse_date_delta, serialize_date_delta, 'HTTP date or delta seconds')

    server = header_getter('Server', '14.38')

    # TODO: the standard allows this to be a list of challenges
    www_authenticate = converter(
        header_getter('WWW-Authenticate', '14.47'),
        parse_auth, serialize_auth,
    )

    #
    # charset
    #

    def _charset__get(self):
        """
        Get/set the ``charset`` specified in ``Content-Type``.

        There is no checking to validate that a ``content_type`` actually
        allows for a ``charset`` parameter.
        """
        header = self.headers.get('Content-Type')
        if not header:
            return None
        match = CHARSET_RE.search(header)
        if match:
            return match.group(1)
        return None

    def _charset__set(self, charset):
        if charset is None:
            self._charset__del()
            return
        header = self.headers.get('Content-Type', None)
        if header is None:
            raise AttributeError("You cannot set the charset when no "
                                 "content-type is defined")
        match = CHARSET_RE.search(header)
        if match:
            header = header[:match.start()] + header[match.end():]
        header += '; charset=%s' % charset
        self.headers['Content-Type'] = header

    def _charset__del(self):
        header = self.headers.pop('Content-Type', None)
        if header is None:
            # Don't need to remove anything
            return
        match = CHARSET_RE.search(header)
        if match:
            header = header[:match.start()] + header[match.end():]
        self.headers['Content-Type'] = header

    charset = property(_charset__get, _charset__set, _charset__del,
                       doc=_charset__get.__doc__)

    #
    # content_type
    #

    def _content_type__get(self):
        """
        Get/set the ``Content-Type`` header. If no ``Content-Type`` header is
        set, this will return ``None``.

        .. versionchanged:: 1.7

            Setting a new ``Content-Type`` will remove all ``Content-Type``
            parameters and reset the ``charset`` to the default if the
            ``Content-Type`` is ``text/*`` or XML (``application/xml`` or
            ``*/*+xml``).

            To preserve all ``Content-Type`` parameters, you may use the
            following code:

            .. code-block:: python

                resp = Response()
                params = resp.content_type_params
                resp.content_type = 'application/something'
                resp.content_type_params = params
        """
        header = self.headers.get('Content-Type')
        if not header:
            return None
        return header.split(';', 1)[0]

    def _content_type__set(self, value):
        if not value:
            self._content_type__del()
            return
        else:
            if PY2 and isinstance(value, text_type):
                value = value.encode("latin-1")

            if not isinstance(value, string_types):
                raise TypeError("content_type requires value to be of string_types")

            content_type = value

            # Set up the charset if the content-type doesn't have one

            has_charset = 'charset=' in content_type

            new_charset = None

            if (
                not has_charset and
                self.default_charset
            ):
                new_charset = self.default_charset

            # Optimize for the default_content_type as shipped by
            # WebOb, becuase we know that 'text/html' has a charset,
            # otherwise add a charset if the content_type has a charset.
            #
            # We add the default charset if the content-type is "texty".
            if (
                new_charset and
                (
                    content_type == 'text/html' or
                    _content_type_has_charset(content_type)
                )
            ):
                content_type += '; charset=' + new_charset

            self.headers['Content-Type'] = content_type

    def _content_type__del(self):
        self.headers.pop('Content-Type', None)

    content_type = property(_content_type__get, _content_type__set,
                            _content_type__del, doc=_content_type__get.__doc__)

    #
    # content_type_params
    #

    def _content_type_params__get(self):
        """
        A dictionary of all the parameters in the content type.

        (This is not a view, set to change, modifications of the dict will not
        be applied otherwise.)
        """
        params = self.headers.get('Content-Type', '')
        if ';' not in params:
            return {}
        params = params.split(';', 1)[1]
        result = {}
        for match in _PARAM_RE.finditer(params):
            result[match.group(1)] = match.group(2) or match.group(3) or ''
        return result

    def _content_type_params__set(self, value_dict):
        if not value_dict:
            self._content_type_params__del()
            return

        params = []
        for k, v in sorted(value_dict.items()):
            if not _OK_PARAM_RE.search(v):
                v = '"%s"' % v.replace('"', '\\"')
            params.append('; %s=%s' % (k, v))
        ct = self.headers.pop('Content-Type', '').split(';', 1)[0]
        ct += ''.join(params)
        self.headers['Content-Type'] = ct

    def _content_type_params__del(self):
        self.headers['Content-Type'] = self.headers.get(
            'Content-Type', '').split(';', 1)[0]

    content_type_params = property(
        _content_type_params__get,
        _content_type_params__set,
        _content_type_params__del,
        _content_type_params__get.__doc__
    )

    #
    # set_cookie, unset_cookie, delete_cookie, merge_cookies
    #

    def set_cookie(self, name, value='', max_age=None,
                   path='/', domain=None, secure=False, httponly=False,
                   comment=None, expires=None, overwrite=False,
                   samesite=None):
        """
        Set (add) a cookie for the response.

        Arguments are:

        ``name``

           The cookie name.

        ``value``

           The cookie value, which should be a string or ``None``.  If
           ``value`` is ``None``, it's equivalent to calling the
           :meth:`webob.response.Response.unset_cookie` method for this
           cookie key (it effectively deletes the cookie on the client).

        ``max_age``

           An integer representing a number of seconds, ``datetime.timedelta``,
           or ``None``. This value is used as the ``Max-Age`` of the generated
           cookie.  If ``expires`` is not passed and this value is not
           ``None``, the ``max_age`` value will also influence the ``Expires``
           value of the cookie (``Expires`` will be set to ``now`` +
           ``max_age``).  If this value is ``None``, the cookie will not have a
           ``Max-Age`` value (unless ``expires`` is set). If both ``max_age``
           and ``expires`` are set, this value takes precedence.

        ``path``

           A string representing the cookie ``Path`` value.  It defaults to
           ``/``.

        ``domain``

           A string representing the cookie ``Domain``, or ``None``.  If
           domain is ``None``, no ``Domain`` value will be sent in the
           cookie.

        ``secure``

           A boolean.  If it's ``True``, the ``secure`` flag will be sent in
           the cookie, if it's ``False``, the ``secure`` flag will not be
           sent in the cookie.

        ``httponly``

           A boolean.  If it's ``True``, the ``HttpOnly`` flag will be sent
           in the cookie, if it's ``False``, the ``HttpOnly`` flag will not
           be sent in the cookie.

        ``samesite``

          A string representing the ``SameSite`` attribute of the cookie or
          ``None``. If samesite is ``None`` no ``SameSite`` value will be sent
          in the cookie. Should only be ``"strict"``, ``"lax"``, or ``"none"``.

        ``comment``

           A string representing the cookie ``Comment`` value, or ``None``.
           If ``comment`` is ``None``, no ``Comment`` value will be sent in
           the cookie.

        ``expires``

           A ``datetime.timedelta`` object representing an amount of time,
           ``datetime.datetime`` or ``None``. A non-``None`` value is used to
           generate the ``Expires`` value of the generated cookie. If
           ``max_age`` is not passed, but this value is not ``None``, it will
           influence the ``Max-Age`` header. If this value is ``None``, the
           ``Expires`` cookie value will be unset (unless ``max_age`` is set).
           If ``max_age`` is set, it will be used to generate the ``expires``
           and this value is ignored.

           If a ``datetime.datetime`` is provided it has to either be timezone
           aware or be based on UTC. ``datetime.datetime`` objects that are
           local time are not supported. Timezone aware ``datetime.datetime``
           objects are converted to UTC.

           This argument will be removed in future versions of WebOb (version
           1.9).

        ``overwrite``

           If this key is ``True``, before setting the cookie, unset any
           existing cookie.

        """

        # Remove in WebOb 1.10
        if expires:
            warn_deprecation('Argument "expires" will be removed in a future '
                             'version of WebOb, please use "max_age".', 1.10, 1)

        if overwrite:
            self.unset_cookie(name, strict=False)

        # If expires is set, but not max_age we set max_age to expires
        if not max_age and isinstance(expires, timedelta):
            max_age = expires

        # expires can also be a datetime
        if not max_age and isinstance(expires, datetime):

            # If expires has a timezone attached, convert it to UTC
            if expires.tzinfo and expires.utcoffset():
                expires = (expires - expires.utcoffset()).replace(tzinfo=None)

            max_age = expires - datetime.utcnow()

        value = bytes_(value, 'utf-8')

        cookie = make_cookie(name, value, max_age=max_age, path=path,
                domain=domain, secure=secure, httponly=httponly,
                comment=comment, samesite=samesite)
        self.headerlist.append(('Set-Cookie', cookie))

    def delete_cookie(self, name, path='/', domain=None):
        """
        Delete a cookie from the client.  Note that ``path`` and ``domain``
        must match how the cookie was originally set.

        This sets the cookie to the empty string, and ``max_age=0`` so
        that it should expire immediately.
        """
        self.set_cookie(name, None, path=path, domain=domain)

    def unset_cookie(self, name, strict=True):
        """
        Unset a cookie with the given name (remove it from the response).
        """
        existing = self.headers.getall('Set-Cookie')
        if not existing and not strict:
            return
        cookies = Cookie()
        for header in existing:
            cookies.load(header)
        if isinstance(name, text_type):
            name = name.encode('utf8')
        if name in cookies:
            del cookies[name]
            del self.headers['Set-Cookie']
            for m in cookies.values():
                self.headerlist.append(('Set-Cookie', m.serialize()))
        elif strict:
            raise KeyError("No cookie has been set with the name %r" % name)

    def merge_cookies(self, resp):
        """Merge the cookies that were set on this response with the
        given ``resp`` object (which can be any WSGI application).

        If the ``resp`` is a :class:`webob.Response` object, then the
        other object will be modified in-place.
        """
        if not self.headers.get('Set-Cookie'):
            return resp
        if isinstance(resp, Response):
            for header in self.headers.getall('Set-Cookie'):
                resp.headers.add('Set-Cookie', header)
            return resp
        else:
            c_headers = [h for h in self.headerlist if
                         h[0].lower() == 'set-cookie']
            def repl_app(environ, start_response):
                def repl_start_response(status, headers, exc_info=None):
                    return start_response(status, headers + c_headers,
                                          exc_info=exc_info)
                return resp(environ, repl_start_response)
            return repl_app

    #
    # cache_control
    #

    _cache_control_obj = None

    def _cache_control__get(self):
        """
        Get/set/modify the Cache-Control header (`HTTP spec section 14.9
        <http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.9>`_).
        """
        value = self.headers.get('cache-control', '')
        if self._cache_control_obj is None:
            self._cache_control_obj = CacheControl.parse(
                value, updates_to=self._update_cache_control, type='response')
            self._cache_control_obj.header_value = value
        if self._cache_control_obj.header_value != value:
            new_obj = CacheControl.parse(value, type='response')
            self._cache_control_obj.properties.clear()
            self._cache_control_obj.properties.update(new_obj.properties)
            self._cache_control_obj.header_value = value
        return self._cache_control_obj

    def _cache_control__set(self, value):
        # This actually becomes a copy
        if not value:
            value = ""
        if isinstance(value, dict):
            value = CacheControl(value, 'response')
        if isinstance(value, text_type):
            value = str(value)
        if isinstance(value, str):
            if self._cache_control_obj is None:
                self.headers['Cache-Control'] = value
                return
            value = CacheControl.parse(value, 'response')
        cache = self.cache_control
        cache.properties.clear()
        cache.properties.update(value.properties)

    def _cache_control__del(self):
        self.cache_control = {}

    def _update_cache_control(self, prop_dict):
        value = serialize_cache_control(prop_dict)
        if not value:
            if 'Cache-Control' in self.headers:
                del self.headers['Cache-Control']
        else:
            self.headers['Cache-Control'] = value

    cache_control = property(
        _cache_control__get, _cache_control__set,
        _cache_control__del, doc=_cache_control__get.__doc__)

    #
    # cache_expires
    #

    def _cache_expires(self, seconds=0, **kw):
        """
            Set expiration on this request.  This sets the response to
            expire in the given seconds, and any other attributes are used
            for ``cache_control`` (e.g., ``private=True``).
        """
        if seconds is True:
            seconds = 0
        elif isinstance(seconds, timedelta):
            seconds = timedelta_to_seconds(seconds)
        cache_control = self.cache_control
        if seconds is None:
            pass
        elif not seconds:
            # To really expire something, you have to force a
            # bunch of these cache control attributes, and IE may
            # not pay attention to those still so we also set
            # Expires.
            cache_control.no_store = True
            cache_control.no_cache = True
            cache_control.must_revalidate = True
            cache_control.max_age = 0
            cache_control.post_check = 0
            cache_control.pre_check = 0
            self.expires = datetime.utcnow()
            if 'last-modified' not in self.headers:
                self.last_modified = datetime.utcnow()
            self.pragma = 'no-cache'
        else:
            cache_control.properties.clear()
            cache_control.max_age = seconds
            self.expires = datetime.utcnow() + timedelta(seconds=seconds)
            self.pragma = None
        for name, value in kw.items():
            setattr(cache_control, name, value)

    cache_expires = property(lambda self: self._cache_expires, _cache_expires)

    #
    # encode_content, decode_content, md5_etag
    #

    def encode_content(self, encoding='gzip', lazy=False):
        """
        Encode the content with the given encoding (only ``gzip`` and
        ``identity`` are supported).
        """
        assert encoding in ('identity', 'gzip'), \
            "Unknown encoding: %r" % encoding
        if encoding == 'identity':
            self.decode_content()
            return
        if self.content_encoding == 'gzip':
            return
        if lazy:
            self.app_iter = gzip_app_iter(self._app_iter)
            self.content_length = None
        else:
            self.app_iter = list(gzip_app_iter(self._app_iter))
            self.content_length = sum(map(len, self._app_iter))
        self.content_encoding = 'gzip'

    def decode_content(self):
        content_encoding = self.content_encoding or 'identity'
        if content_encoding == 'identity':
            return
        if content_encoding not in ('gzip', 'deflate'):
            raise ValueError(
                "I don't know how to decode the content %s" % content_encoding)
        if content_encoding == 'gzip':
            from gzip import GzipFile
            from io import BytesIO
            gzip_f = GzipFile(filename='', mode='r', fileobj=BytesIO(self.body))
            self.body = gzip_f.read()
            self.content_encoding = None
            gzip_f.close()
        else:
            try:
                # RFC7230 section 4.2.2 specifies that the body should be wrapped
                # inside a ZLIB (RFC1950) container ...
                self.body = zlib.decompress(self.body)
            except zlib.error:
                # ... but there are nonconformant implementations around which send
                # the data without the ZLIB container, so we use maximum window size
                # decompression without header check (the - sign)
                self.body = zlib.decompress(self.body, -15)
            self.content_encoding = None

    def md5_etag(self, body=None, set_content_md5=False):
        """
        Generate an etag for the response object using an MD5 hash of
        the body (the ``body`` parameter, or ``self.body`` if not given).

        Sets ``self.etag``.

        If ``set_content_md5`` is ``True``, sets ``self.content_md5`` as well.
        """
        if body is None:
            body = self.body
        md5_digest = md5(body).digest()
        md5_digest = b64encode(md5_digest)
        md5_digest = md5_digest.replace(b'\n', b'')
        md5_digest = native_(md5_digest)
        self.etag = md5_digest.strip('=')
        if set_content_md5:
            self.content_md5 = md5_digest

    @staticmethod
    def _make_location_absolute(environ, value):
        if SCHEME_RE.search(value):
            return value

        new_location = urlparse.urljoin(_request_uri(environ), value)
        return new_location

    def _abs_headerlist(self, environ):
        # Build the headerlist, if we have a Location header, make it absolute
        return [
            (k, v) if k.lower() != 'location'
            else (k, self._make_location_absolute(environ, v))
            for (k, v)
            in self._headerlist
        ]

    #
    # __call__, conditional_response_app
    #

    def __call__(self, environ, start_response):
        """
        WSGI application interface
        """
        if self.conditional_response:
            return self.conditional_response_app(environ, start_response)

        headerlist = self._abs_headerlist(environ)

        start_response(self.status, headerlist)
        if environ['REQUEST_METHOD'] == 'HEAD':
            # Special case here...
            return EmptyResponse(self._app_iter)
        return self._app_iter

    _safe_methods = ('GET', 'HEAD')

    def conditional_response_app(self, environ, start_response):
        """
        Like the normal ``__call__`` interface, but checks conditional headers:

            * ``If-Modified-Since``   (``304 Not Modified``; only on ``GET``,
              ``HEAD``)
            * ``If-None-Match``       (``304 Not Modified``; only on ``GET``,
              ``HEAD``)
            * ``Range``               (``406 Partial Content``; only on ``GET``,
              ``HEAD``)
        """
        req = BaseRequest(environ)

        headerlist = self._abs_headerlist(environ)

        method = environ.get('REQUEST_METHOD', 'GET')
        if method in self._safe_methods:
            status304 = False
            if req.if_none_match and self.etag:
                status304 = self.etag in req.if_none_match
            elif req.if_modified_since and self.last_modified:
                status304 = self.last_modified <= req.if_modified_since
            if status304:
                start_response('304 Not Modified', filter_headers(headerlist))
                return EmptyResponse(self._app_iter)
        if (
            req.range and self in req.if_range and
            self.content_range is None and
            method in ('HEAD', 'GET') and
            self.status_code == 200 and
            self.content_length is not None
        ):
            content_range = req.range.content_range(self.content_length)
            if content_range is None:
                iter_close(self._app_iter)
                body = bytes_("Requested range not satisfiable: %s" % req.range)
                headerlist = [
                    ('Content-Length', str(len(body))),
                    ('Content-Range', str(ContentRange(None, None,
                                                       self.content_length))),
                    ('Content-Type', 'text/plain'),
                ] + filter_headers(headerlist)
                start_response('416 Requested Range Not Satisfiable',
                               headerlist)
                if method == 'HEAD':
                    return ()
                return [body]
            else:
                app_iter = self.app_iter_range(content_range.start,
                                               content_range.stop)
                if app_iter is not None:
                    # the following should be guaranteed by
                    # Range.range_for_length(length)
                    assert content_range.start is not None
                    headerlist = [
                        ('Content-Length',
                         str(content_range.stop - content_range.start)),
                        ('Content-Range', str(content_range)),
                    ] + filter_headers(headerlist, ('content-length',))
                    start_response('206 Partial Content', headerlist)
                    if method == 'HEAD':
                        return EmptyResponse(app_iter)
                    return app_iter

        start_response(self.status, headerlist)
        if method == 'HEAD':
            return EmptyResponse(self._app_iter)
        return self._app_iter

    def app_iter_range(self, start, stop):
        """
        Return a new ``app_iter`` built from the response ``app_iter``, that
        serves up only the given ``start:stop`` range.
        """
        app_iter = self._app_iter
        if hasattr(app_iter, 'app_iter_range'):
            return app_iter.app_iter_range(start, stop)
        return AppIterRange(app_iter, start, stop)


def filter_headers(hlist, remove_headers=('content-length', 'content-type')):
    return [h for h in hlist if (h[0].lower() not in remove_headers)]


def iter_file(file, block_size=1 << 18): # 256Kb
    while True:
        data = file.read(block_size)
        if not data:
            break
        yield data

class ResponseBodyFile(object):
    mode = 'wb'
    closed = False

    def __init__(self, response):
        """
        Represents a :class:`~Response` as a file like object.
        """
        self.response = response
        self.write = response.write

    def __repr__(self):
        return '<body_file for %r>' % self.response

    encoding = property(
        lambda self: self.response.charset,
        doc="The encoding of the file (inherited from response.charset)"
    )

    def writelines(self, seq):
        """
        Write a sequence of lines to the response.
        """
        for item in seq:
            self.write(item)

    def close(self):
        raise NotImplementedError("Response bodies cannot be closed")

    def flush(self):
        pass

    def tell(self):
        """
        Provide the current location where we are going to start writing.
        """
        if not self.response.has_body:
            return 0

        return sum([len(chunk) for chunk in self.response.app_iter])


class AppIterRange(object):
    """
    Wraps an ``app_iter``, returning just a range of bytes.
    """

    def __init__(self, app_iter, start, stop):
        assert start >= 0, "Bad start: %r" % start
        assert stop is None or (stop >= 0 and stop >= start), (
            "Bad stop: %r" % stop)
        self.app_iter = iter(app_iter)
        self._pos = 0 # position in app_iter
        self.start = start
        self.stop = stop

    def __iter__(self):
        return self

    def _skip_start(self):
        start, stop = self.start, self.stop
        for chunk in self.app_iter:
            self._pos += len(chunk)
            if self._pos < start:
                continue
            elif self._pos == start:
                return b''
            else:
                chunk = chunk[start - self._pos:]
                if stop is not None and self._pos > stop:
                    chunk = chunk[:stop - self._pos]
                    assert len(chunk) == stop - start
                return chunk
        else:
            raise StopIteration()

    def next(self):
        if self._pos < self.start:
            # need to skip some leading bytes
            return self._skip_start()
        stop = self.stop
        if stop is not None and self._pos >= stop:
            raise StopIteration

        chunk = next(self.app_iter)
        self._pos += len(chunk)

        if stop is None or self._pos <= stop:
            return chunk
        else:
            return chunk[:stop - self._pos]

    __next__ = next # py3

    def close(self):
        iter_close(self.app_iter)


class EmptyResponse(object):
    """
    An empty WSGI response.

    An iterator that immediately stops. Optionally provides a close
    method to close an underlying ``app_iter`` it replaces.
    """

    def __init__(self, app_iter=None):
        if app_iter is not None and hasattr(app_iter, 'close'):
            self.close = app_iter.close

    def __iter__(self):
        return self

    def __len__(self):
        return 0

    def next(self):
        raise StopIteration()

    __next__ = next # py3

def _is_xml(content_type):
    return (
        content_type.startswith('application/xml') or
        (
            content_type.startswith('application/') and
            content_type.endswith('+xml')
        ) or
        (
            content_type.startswith('image/') and
            content_type.endswith('+xml')
        )
    )

def _content_type_has_charset(content_type):
    return (
        content_type.startswith('text/') or
        _is_xml(content_type)
    )

def _request_uri(environ):
    """Like ``wsgiref.url.request_uri``, except eliminates ``:80`` ports.

    Returns the full request URI."""
    url = environ['wsgi.url_scheme'] + '://'

    if environ.get('HTTP_HOST'):
        url += environ['HTTP_HOST']
    else:
        url += environ['SERVER_NAME'] + ':' + environ['SERVER_PORT']
    if url.endswith(':80') and environ['wsgi.url_scheme'] == 'http':
        url = url[:-3]
    elif url.endswith(':443') and environ['wsgi.url_scheme'] == 'https':
        url = url[:-4]

    if PY2:
        script_name = environ.get('SCRIPT_NAME', '/')
        path_info = environ.get('PATH_INFO', '')
    else:
        script_name = bytes_(environ.get('SCRIPT_NAME', '/'), 'latin-1')
        path_info = bytes_(environ.get('PATH_INFO', ''), 'latin-1')

    url += url_quote(script_name)
    qpath_info = url_quote(path_info)
    if 'SCRIPT_NAME' not in environ:
        url += qpath_info[1:]
    else:
        url += qpath_info
    return url


def iter_close(iter):
    if hasattr(iter, 'close'):
        iter.close()

def gzip_app_iter(app_iter):
    size = 0
    crc = zlib.crc32(b"") & 0xffffffff
    compress = zlib.compressobj(9, zlib.DEFLATED, -zlib.MAX_WBITS,
                                zlib.DEF_MEM_LEVEL, 0)

    yield _gzip_header
    for item in app_iter:
        size += len(item)
        crc = zlib.crc32(item, crc) & 0xffffffff

        # The compress function may return zero length bytes if the input is
        # small enough; it buffers the input for the next iteration or for a
        # flush.
        result = compress.compress(item)
        if result:
            yield result

    # Similarly, flush may also not yield a value.
    result = compress.flush()
    if result:
        yield result
    yield struct.pack("<2L", crc, size & 0xffffffff)

def _error_unicode_in_app_iter(app_iter, body):
    app_iter_repr = repr(app_iter)
    if len(app_iter_repr) > 50:
        app_iter_repr = (
            app_iter_repr[:30] + '...' + app_iter_repr[-10:])
    raise TypeError(
        'An item of the app_iter (%s) was text, causing a '
        'text body: %r' % (app_iter_repr, body))