1617 lines (1336 sloc) 53.7 KB
from base64 import b64encode
from datetime import (
from hashlib import md5
import re
import struct
import zlib
import simplejson as json
except ImportError:
import json
from webob.byterange import ContentRange
from webob.cachecontrol import (
from webob.compat import (
from webob.cookies import (
from webob.datetime_utils import (
from webob.descriptors import (
from webob.headers import ResponseHeaders
from webob.request import BaseRequest
from webob.util import status_reasons, status_generic_reasons, warn_deprecation
__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
: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
.. _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
* 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``
* The ``unicode_errors`` is set to ``strict``, and access on a
:attr:`~Response.text` will raise an error if it fails to decode the
* ``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,
# 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')
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'
self.status = status
# Initialize headers
self._headers = None
if headerlist is None:
self._headerlist = []
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
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 += '; 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
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 "
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)
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
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/'
_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
header_name, value = line.split(_colon, 1)
except ValueError:
raise ValueError('Bad header line: %r' % line)
value = value.strip()
native_(header_name, 'latin-1'),
native_(value, 'latin-1')
r = cls(
body = or 0)
if is_text:
r.text = body
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)
# and this to make sure app_iter instances are different
self._app_iter = list(app_iter)
return self.__class__(
# __repr__, __str__
def __repr__(self):
return '<%s at 0x%x %s>' % (self.__class__.__name__, abs(id(self)),
def __str__(self, skip_body=False):
parts = [self.status]
if not skip_body:
# Force enumeration of the body (to set content-length)
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):
code = int(value)
except (ValueError, TypeError):
self.status_code = code
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
# 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
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):
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,
# 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")
body = b''.join(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
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 "
% (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)")
msg = ("You can only set the body to a binary type (not %s)" %
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
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,
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):
new_app_iter = self._app_iter = list(app_iter)
app_iter = new_app_iter
self.content_length = sum(len(chunk) for chunk in app_iter)
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.content_md5 = 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,
# 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(
parse_etag_response, serialize_etag_response,
'Entity tag'
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 =
if match:
return None
def _charset__set(self, charset):
if charset is None:
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 =
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
match =
if match:
header = header[:match.start()] + header[match.end():]
self.headers['Content-Type'] = header
charset = property(_charset__get, _charset__set, _charset__del,
# 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
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:
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
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 += '; 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[] = or or ''
return result
def _content_type_params__set(self, value_dict):
if not value_dict:
params = []
for k, v in sorted(value_dict.items()):
if not
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(
# set_cookie, unset_cookie, delete_cookie, merge_cookies
def set_cookie(self, name=None, value='', max_age=None,
path='/', domain=None, secure=False, httponly=False,
comment=None, expires=None, overwrite=False):
Set (add) a cookie for the response.
Arguments are:
The cookie name.
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).
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.
A string representing the cookie ``Path`` value. It defaults to
A string representing the cookie ``Domain``, or ``None``. If
domain is ``None``, no ``Domain`` value will be sent in the
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.
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.
A string representing the cookie ``Comment`` value, or ``None``.
If ``comment`` is ``None``, no ``Comment`` value will be sent in
the cookie.
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
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 name is None:
raise TypeError('set_cookie() takes at least 1 argument')
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,
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:
cookies = Cookie()
for header in existing:
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
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,
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
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.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
value = CacheControl.parse(value, 'response')
cache = self.cache_control
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']
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:
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'
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':
if self.content_encoding == 'gzip':
if lazy:
self.app_iter = gzip_app_iter(self._app_iter)
self.content_length = None
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':
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 =
self.content_encoding = None
# Weird feature:
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
def _make_location_absolute(environ, 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``,
* ``If-None-Match`` (``304 Not Modified``; only on ``GET``,
* ``Range`` (``406 Partial Content``; only on ``GET``,
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:
body = bytes_("Requested range not satisfiable: %s" % req.range)
headerlist = [
('Content-Length', str(len(body))),
('Content-Range', str(ContentRange(None, None,
('Content-Type', 'text/plain'),
] + filter_headers(headerlist)
start_response('416 Requested Range Not Satisfiable',
if method == 'HEAD':
return ()
return [body]
app_iter = self.app_iter_range(content_range.start,
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 = [
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 =
if not data:
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:
def close(self):
raise NotImplementedError("Response bodies cannot be closed")
def flush(self):
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:
elif self._pos == start:
return b''
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
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
return chunk[:stop - self._pos]
__next__ = next # py3
def close(self):
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
) or
content_type.startswith('image/') and
def _content_type_has_charset(content_type):
return (
content_type.startswith('text/') or
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']
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', '')
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:]
url += qpath_info
return url
def iter_close(iter):
if hasattr(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))