| from base64 import b64encode |
| from datetime import ( |
| datetime, |
| timedelta, |
| ) |
| from hashlib import md5 |
| import re |
| import struct |
| import zlib |
| try: |
| import simplejson as json |
| except ImportError: |
| import json |
| |
| from webob.byterange import ContentRange |
| |
| from webob.cachecontrol import ( |
| CacheControl, |
| serialize_cache_control, |
| ) |
| |
| from webob.compat import ( |
| PY3, |
| bytes_, |
| native_, |
| 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_reasons, status_generic_reasons |
| |
| __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' |
| |
| class Response(object): |
| """ |
| Represents a WSGI response |
| """ |
| |
| default_content_type = 'text/html' |
| default_charset = 'UTF-8' # TODO: deprecate |
| unicode_errors = 'strict' # TODO: deprecate (why would response body have errors?) |
| default_conditional_response = False |
| 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, |
| **kw): |
| 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=(',', ':')) |
| 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") |
| if status is None: |
| self._status = '200 OK' |
| else: |
| self.status = status |
| if headerlist is None: |
| self._headerlist = [] |
| else: |
| self._headerlist = headerlist |
| self._headers = None |
| if content_type is None: |
| content_type = self.default_content_type |
| charset = None |
| if 'charset' in kw: |
| charset = kw.pop('charset') |
| elif self.default_charset: |
| if (content_type |
| and 'charset=' not in content_type |
| and (content_type == 'text/html' |
| or content_type.startswith('text/') |
| or content_type.startswith('application/xml') |
| or content_type.startswith('application/json') |
| or (content_type.startswith('application/') |
| and (content_type.endswith('+xml') or content_type.endswith('+json'))))): |
| charset = self.default_charset |
| if content_type and charset: |
| content_type += '; charset=' + charset |
| elif self._headerlist and charset: |
| self.charset = charset |
| if not self._headerlist and content_type: |
| self._headerlist.append(('Content-Type', content_type)) |
| if conditional_response is None: |
| self.conditional_response = self.default_conditional_response |
| else: |
| self.conditional_response = bool(conditional_response) |
| if app_iter is None: |
| if isinstance(body, text_type): |
| if charset is None: |
| raise TypeError( |
| "You cannot set the body to a text value without a " |
| "charset") |
| body = body.encode(charset) |
| app_iter = [body] |
| if headerlist is None: |
| self._headerlist.append(('Content-Length', str(len(body)))) |
| else: |
| self.headers['Content-Length'] = str(len(body)) |
| self._app_iter = app_iter |
| 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 = ':' |
| else: |
| _colon = b':' |
| 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__( |
| content_type=False, |
| 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.text if PY3 else self.body] |
| 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 PY3: # pragma: no cover |
| 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 |
| status_code = int(value.split()[0]) |
| 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 ``str``. 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 # pragma: no cover (no idea why necessary, it's hit) |
| 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): |
| """Access the body of the response as JSON""" |
| # Note: UTF-8 is a content-type specific default for JSON: |
| return json.loads(self.body.decode(self.charset or 'UTF-8')) |
| |
| def _json_body__set(self, value): |
| self.body = json.dumps(value, separators=(',', ':')).encode(self.charset or 'UTF-8') |
| |
| def _json_body__del(self): |
| del self.body |
| |
| json = json_body = property(_json_body__get, _json_body__set, _json_body__del) |
| |
| |
| # |
| # text, unicode_body, ubody |
| # |
| |
| def _text__get(self): |
| """ |
| Get/set the text value of the body (using the charset of the |
| Content-Type) |
| """ |
| if not self.charset: |
| raise AttributeError( |
| "You cannot access Response.text unless charset is set") |
| body = self.body |
| return body.decode(self.charset, self.unicode_errors) |
| |
| def _text__set(self, value): |
| if not self.charset: |
| raise AttributeError( |
| "You cannot access Response.text unless charset is set") |
| if not isinstance(value, text_type): |
| raise TypeError( |
| "You can only set Response.text to a unicode string " |
| "(not %s)" % type(value)) |
| self.body = value.encode(self.charset) |
| |
| 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.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, |
| 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 (in the Content-Type) |
| """ |
| 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: |
| del self.charset |
| return |
| header = self.headers.pop('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 (or None), *without* the |
| charset or any parameters. |
| |
| If you include parameters (or ``;`` at all) when setting the |
| content_type, any existing parameters will be deleted; |
| otherwise they will be preserved. |
| """ |
| 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 |
| if ';' not in value: |
| header = self.headers.get('Content-Type', '') |
| if ';' in header: |
| params = header.split(';', 1)[1] |
| value += ';' + params |
| self.headers['Content-Type'] = value |
| |
| 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 would 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: |
| del self.content_type_params |
| 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): |
| """ |
| 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. |
| |
| ``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. |
| |
| ``overwrite`` |
| |
| If this key is ``True``, before setting the cookie, unset any |
| existing cookie. |
| |
| """ |
| 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): |
| 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) |
| |
| 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, etc). |
| """ |
| 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: |
| # Weird feature: http://bugs.python.org/issue5784 |
| 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 |
| |
| |
| |
| # |
| # __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 |
| |
| def _abs_headerlist(self, environ): |
| """Returns a headerlist, with the Location header possibly |
| made absolute given the request environ. |
| """ |
| headerlist = list(self.headerlist) |
| for i, (name, value) in enumerate(headerlist): |
| if name.lower() == 'location': |
| if SCHEME_RE.search(value): |
| break |
| new_location = urlparse.urljoin(_request_uri(environ), value) |
| headerlist[i] = (name, new_location) |
| break |
| return headerlist |
| |
| _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): |
| 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): |
| for item in seq: |
| self.write(item) |
| |
| def close(self): |
| raise NotImplementedError("Response bodies cannot be closed") |
| |
| def flush(self): |
| pass |
| |
| |
| |
| 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 _request_uri(environ): |
| """Like wsgiref.url.request_uri, except eliminates :80 ports |
| |
| Return 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 PY3: # pragma: no cover |
| script_name = bytes_(environ.get('SCRIPT_NAME', '/'), 'latin-1') |
| path_info = bytes_(environ.get('PATH_INFO', ''), 'latin-1') |
| else: |
| script_name = environ.get('SCRIPT_NAME', '/') |
| path_info = environ.get('PATH_INFO', '') |
| |
| url += url_quote(script_name) |
| qpath_info = url_quote(path_info) |
| if not 'SCRIPT_NAME' 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)) |
