<s> """<STR_LIT>""" <EOL> import urlparse <EOL> from datetime import datetime , timedelta <EOL> from werkzeug . http import HTTP_STATUS_CODES , parse_accept_header , parse_cache_control_header , parse_etags , parse_date , generate_etag , is_resource_modified , unquote_etag , quote_etag , parse_set_header , parse_authorization_header , parse_www_authenticate_header , remove_entity_headers , parse_options_header , dump_options_header , http_date , parse_if_range_header , parse_cookie , dump_cookie , parse_range_header , parse_content_range_header , dump_header <EOL> from werkzeug . urls import url_decode , iri_to_uri <EOL> from werkzeug . formparser import FormDataParser , default_stream_factory <EOL> from werkzeug . utils import cached_property , environ_property , header_property , get_content_type <EOL> from werkzeug . wsgi import get_current_url , get_host , LimitedStream , ClosingIterator <EOL> from werkzeug . datastructures import MultiDict , CombinedMultiDict , Headers , EnvironHeaders , ImmutableMultiDict , ImmutableTypeConversionDict , ImmutableList , MIMEAccept , CharsetAccept , LanguageAccept , ResponseCacheControl , RequestCacheControl , CallbackDict , ContentRange <EOL> from werkzeug . _internal import _empty_stream , _decode_unicode , _patch_wrapper , _get_environ <EOL> def _run_wsgi_app ( * args ) : <EOL> """<STR_LIT>""" <EOL> global _run_wsgi_app <EOL> from werkzeug . test import run_wsgi_app as _run_wsgi_app <EOL> return _run_wsgi_app ( * args ) <EOL> def _warn_if_string ( iterable ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( iterable , basestring ) : <EOL> from warnings import warn <EOL> warn ( Warning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , stacklevel = <NUM_LIT:2> ) <EOL> class BaseRequest ( object ) : <EOL> """<STR_LIT>""" <EOL> charset = '<STR_LIT:utf-8>' <EOL> encoding_errors = '<STR_LIT:replace>' <EOL> max_content_length = None <EOL> max_form_memory_size = None <EOL> parameter_storage_class = ImmutableMultiDict <EOL> list_storage_class = ImmutableList <EOL> dict_storage_class = ImmutableTypeConversionDict <EOL> form_data_parser_class = FormDataParser <EOL> def __init__ ( self , environ , populate_request = True , shallow = False ) : <EOL> self . environ = environ <EOL> if populate_request and not shallow : <EOL> self . environ [ '<STR_LIT>' ] = self <EOL> self . shallow = shallow <EOL> def __repr__ ( self ) : <EOL> args = [ ] <EOL> try : <EOL> args . append ( "<STR_LIT>" % self . url ) <EOL> args . append ( '<STR_LIT>' % self . method ) <EOL> except Exception : <EOL> args . append ( '<STR_LIT>' ) <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , <EOL> '<STR_LIT:U+0020>' . join ( args ) <EOL> ) <EOL> @ property <EOL> def url_charset ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . charset <EOL> @ classmethod <EOL> def from_values ( cls , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> from werkzeug . test import EnvironBuilder <EOL> charset = kwargs . pop ( '<STR_LIT>' , cls . charset ) <EOL> builder = EnvironBuilder ( * args , ** kwargs ) <EOL> try : <EOL> return builder . get_request ( cls ) <EOL> finally : <EOL> builder . close ( ) <EOL> @ classmethod <EOL> def application ( cls , f ) : <EOL> """<STR_LIT>""" <EOL> return _patch_wrapper ( f , lambda * a : f ( * a [ : - <NUM_LIT:2> ] + ( cls ( a [ - <NUM_LIT:2> ] ) , ) ) ( * a [ - <NUM_LIT:2> : ] ) ) <EOL> def _get_file_stream ( self , total_content_length , content_type , filename = None , <EOL> content_length = None ) : <EOL> """<STR_LIT>""" <EOL> return default_stream_factory ( total_content_length , content_type , <EOL> filename , content_length ) <EOL> @ property <EOL> def want_form_data_parsed ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . environ [ '<STR_LIT>' ] in ( '<STR_LIT:POST>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> def make_form_data_parser ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . form_data_parser_class ( self . _get_file_stream , <EOL> self . charset , <EOL> self . encoding_errors , <EOL> self . max_form_memory_size , <EOL> self . max_content_length , <EOL> self . parameter_storage_class ) <EOL> def _load_form_data ( self ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in self . __dict__ : <EOL> return <EOL> if self . shallow : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> data = None <EOL> stream = _empty_stream <EOL> if self . want_form_data_parsed : <EOL> parser = self . make_form_data_parser ( ) <EOL> data = parser . parse_from_environ ( self . environ ) <EOL> else : <EOL> content_length = self . headers . get ( '<STR_LIT>' , type = int ) <EOL> if content_length is not None : <EOL> stream = LimitedStream ( self . environ [ '<STR_LIT>' ] , <EOL> content_length ) <EOL> if data is None : <EOL> data = ( stream , self . parameter_storage_class ( ) , <EOL> self . parameter_storage_class ( ) ) <EOL> d = self . __dict__ <EOL> d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] , d [ '<STR_LIT>' ] = data <EOL> @ cached_property <EOL> def stream ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _load_form_data ( ) <EOL> return self . stream <EOL> input_stream = environ_property ( '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ cached_property <EOL> def args ( self ) : <EOL> """<STR_LIT>""" <EOL> return url_decode ( self . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> self . url_charset , errors = self . encoding_errors , <EOL> cls = self . parameter_storage_class ) <EOL> @ cached_property <EOL> def data ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . stream . read ( ) <EOL> @ cached_property <EOL> def form ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _load_form_data ( ) <EOL> return self . form <EOL> @ cached_property <EOL> def values ( self ) : <EOL> """<STR_LIT>""" <EOL> args = [ ] <EOL> for d in self . args , self . form : <EOL> if not isinstance ( d , MultiDict ) : <EOL> d = MultiDict ( d ) <EOL> args . append ( d ) <EOL> return CombinedMultiDict ( args ) <EOL> @ cached_property <EOL> def files ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _load_form_data ( ) <EOL> return self . files <EOL> @ cached_property <EOL> def cookies ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_cookie ( self . environ , self . charset , <EOL> cls = self . dict_storage_class ) <EOL> @ cached_property <EOL> def headers ( self ) : <EOL> """<STR_LIT>""" <EOL> return EnvironHeaders ( self . environ ) <EOL> @ cached_property <EOL> def path ( self ) : <EOL> """<STR_LIT>""" <EOL> path = '<STR_LIT:/>' + ( self . environ . get ( '<STR_LIT>' ) or '<STR_LIT>' ) . lstrip ( '<STR_LIT:/>' ) <EOL> return _decode_unicode ( path , self . url_charset , self . encoding_errors ) <EOL> @ cached_property <EOL> def script_root ( self ) : <EOL> """<STR_LIT>""" <EOL> path = ( self . environ . get ( '<STR_LIT>' ) or '<STR_LIT>' ) . rstrip ( '<STR_LIT:/>' ) <EOL> return _decode_unicode ( path , self . url_charset , self . encoding_errors ) <EOL> @ cached_property <EOL> def url ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_current_url ( self . environ ) <EOL> @ cached_property <EOL> def base_url ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_current_url ( self . environ , strip_querystring = True ) <EOL> @ cached_property <EOL> def url_root ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_current_url ( self . environ , True ) <EOL> @ cached_property <EOL> def host_url ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_current_url ( self . environ , host_only = True ) <EOL> @ cached_property <EOL> def host ( self ) : <EOL> """<STR_LIT>""" <EOL> return get_host ( self . environ ) <EOL> query_string = environ_property ( '<STR_LIT>' , '<STR_LIT>' , read_only = True , doc = <EOL> '''<STR_LIT>''' ) <EOL> method = environ_property ( '<STR_LIT>' , '<STR_LIT:GET>' , read_only = True , doc = <EOL> '''<STR_LIT>''' ) <EOL> @ cached_property <EOL> def access_route ( self ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT>' in self . environ : <EOL> addr = self . environ [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) <EOL> return self . list_storage_class ( [ x . strip ( ) for x in addr ] ) <EOL> elif '<STR_LIT>' in self . environ : <EOL> return self . list_storage_class ( [ self . environ [ '<STR_LIT>' ] ] ) <EOL> return self . list_storage_class ( ) <EOL> @ property <EOL> def remote_addr ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . environ . get ( '<STR_LIT>' ) <EOL> remote_user = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> scheme = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> is_xhr = property ( lambda x : x . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> . lower ( ) == '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> is_secure = property ( lambda x : x . environ [ '<STR_LIT>' ] == '<STR_LIT>' , <EOL> doc = '<STR_LIT>' ) <EOL> is_multithread = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> is_multiprocess = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> is_run_once = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> class BaseResponse ( object ) : <EOL> """<STR_LIT>""" <EOL> charset = '<STR_LIT:utf-8>' <EOL> default_status = <NUM_LIT:200> <EOL> default_mimetype = '<STR_LIT>' <EOL> implicit_sequence_conversion = True <EOL> autocorrect_location_header = True <EOL> automatically_set_content_length = True <EOL> def __init__ ( self , response = None , status = None , headers = None , <EOL> mimetype = None , content_type = None , direct_passthrough = False ) : <EOL> if isinstance ( headers , Headers ) : <EOL> self . headers = headers <EOL> elif not headers : <EOL> self . headers = Headers ( ) <EOL> else : <EOL> self . headers = Headers ( headers ) <EOL> if content_type is None : <EOL> if mimetype is None and '<STR_LIT>' not in self . headers : <EOL> mimetype = self . default_mimetype <EOL> if mimetype is not None : <EOL> mimetype = get_content_type ( mimetype , self . charset ) <EOL> content_type = mimetype <EOL> if content_type is not None : <EOL> self . headers [ '<STR_LIT:Content-Type>' ] = content_type <EOL> if status is None : <EOL> status = self . default_status <EOL> if isinstance ( status , ( int , long ) ) : <EOL> self . status_code = status <EOL> else : <EOL> self . status = status <EOL> self . direct_passthrough = direct_passthrough <EOL> self . _on_close = [ ] <EOL> if response is None : <EOL> self . response = [ ] <EOL> elif isinstance ( response , basestring ) : <EOL> self . data = response <EOL> else : <EOL> self . response = response <EOL> def call_on_close ( self , func ) : <EOL> """<STR_LIT>""" <EOL> self . _on_close . append ( func ) <EOL> return func <EOL> def __repr__ ( self ) : <EOL> if self . is_sequence : <EOL> body_info = '<STR_LIT>' % sum ( map ( len , self . iter_encoded ( ) ) ) <EOL> else : <EOL> body_info = self . is_streamed and '<STR_LIT>' or '<STR_LIT>' <EOL> return '<STR_LIT>' % ( <EOL> self . __class__ . __name__ , <EOL> body_info , <EOL> self . status <EOL> ) <EOL> @ classmethod <EOL> def force_type ( cls , response , environ = None ) : <EOL> """<STR_LIT>""" <EOL> if not isinstance ( response , BaseResponse ) : <EOL> if environ is None : <EOL> raise TypeError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> response = BaseResponse ( * _run_wsgi_app ( response , environ ) ) <EOL> response . __class__ = cls <EOL> return response <EOL> @ classmethod <EOL> def from_app ( cls , app , environ , buffered = False ) : <EOL> """<STR_LIT>""" <EOL> return cls ( * _run_wsgi_app ( app , environ , buffered ) ) <EOL> def _get_status_code ( self ) : <EOL> return self . _status_code <EOL> def _set_status_code ( self , code ) : <EOL> self . _status_code = code <EOL> try : <EOL> self . _status = '<STR_LIT>' % ( code , HTTP_STATUS_CODES [ code ] . upper ( ) ) <EOL> except KeyError : <EOL> self . _status = '<STR_LIT>' % code <EOL> status_code = property ( _get_status_code , _set_status_code , <EOL> doc = '<STR_LIT>' ) <EOL> del _get_status_code , _set_status_code <EOL> def _get_status ( self ) : <EOL> return self . _status <EOL> def _set_status ( self , value ) : <EOL> self . _status = value <EOL> try : <EOL> self . _status_code = int ( self . _status . split ( None , <NUM_LIT:1> ) [ <NUM_LIT:0> ] ) <EOL> except ValueError : <EOL> self . _status_code = <NUM_LIT:0> <EOL> status = property ( _get_status , _set_status , doc = '<STR_LIT>' ) <EOL> del _get_status , _set_status <EOL> def _get_data ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _ensure_sequence ( ) <EOL> return '<STR_LIT>' . join ( self . iter_encoded ( ) ) <EOL> def _set_data ( self , value ) : <EOL> if isinstance ( value , unicode ) : <EOL> value = value . encode ( self . charset ) <EOL> self . response = [ value ] <EOL> if self . automatically_set_content_length : <EOL> self . headers [ '<STR_LIT>' ] = str ( len ( value ) ) <EOL> data = property ( _get_data , _set_data , doc = _get_data . __doc__ ) <EOL> del _get_data , _set_data <EOL> def _ensure_sequence ( self , mutable = False ) : <EOL> """<STR_LIT>""" <EOL> if self . is_sequence : <EOL> if mutable and not isinstance ( self . response , list ) : <EOL> self . response = list ( self . response ) <EOL> return <EOL> if not self . implicit_sequence_conversion : <EOL> raise RuntimeError ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . make_sequence ( ) <EOL> def make_sequence ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . is_sequence : <EOL> close = getattr ( self . response , '<STR_LIT>' , None ) <EOL> self . response = list ( self . iter_encoded ( ) ) <EOL> if close is not None : <EOL> self . call_on_close ( close ) <EOL> def iter_encoded ( self , charset = None ) : <EOL> """<STR_LIT>""" <EOL> if __debug__ and charset is not None : <EOL> from warnings import warn <EOL> warn ( DeprecationWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> charset = self . charset <EOL> if __debug__ : <EOL> _warn_if_string ( self . response ) <EOL> for item in self . response : <EOL> if isinstance ( item , unicode ) : <EOL> yield item . encode ( charset ) <EOL> else : <EOL> yield str ( item ) <EOL> def set_cookie ( self , key , value = '<STR_LIT>' , max_age = None , expires = None , <EOL> path = '<STR_LIT:/>' , domain = None , secure = None , httponly = False ) : <EOL> """<STR_LIT>""" <EOL> self . headers . add ( '<STR_LIT>' , dump_cookie ( key , value , max_age , <EOL> expires , path , domain , secure , httponly , <EOL> self . charset ) ) <EOL> def delete_cookie ( self , key , path = '<STR_LIT:/>' , domain = None ) : <EOL> """<STR_LIT>""" <EOL> self . set_cookie ( key , expires = <NUM_LIT:0> , max_age = <NUM_LIT:0> , path = path , domain = domain ) <EOL> @ property <EOL> def header_list ( self ) : <EOL> if __debug__ : <EOL> from warnings import warn <EOL> warn ( DeprecationWarning ( '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> return self . headers . to_list ( self . charset ) <EOL> @ property <EOL> def is_streamed ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> len ( self . response ) <EOL> except TypeError : <EOL> return True <EOL> return False <EOL> @ property <EOL> def is_sequence ( self ) : <EOL> """<STR_LIT>""" <EOL> return isinstance ( self . response , ( tuple , list ) ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if hasattr ( self . response , '<STR_LIT>' ) : <EOL> self . response . close ( ) <EOL> for func in self . _on_close : <EOL> func ( ) <EOL> def freeze ( self ) : <EOL> """<STR_LIT>""" <EOL> self . response = list ( self . iter_encoded ( ) ) <EOL> self . headers [ '<STR_LIT>' ] = str ( sum ( map ( len , self . response ) ) ) <EOL> def fix_headers ( self , environ ) : <EOL> if __debug__ : <EOL> from warnings import warn <EOL> warn ( DeprecationWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> self . headers [ : ] = self . get_wsgi_headers ( environ ) <EOL> def get_wsgi_headers ( self , environ ) : <EOL> """<STR_LIT>""" <EOL> headers = Headers ( self . headers ) <EOL> location = None <EOL> content_location = None <EOL> content_length = None <EOL> status = self . status_code <EOL> for key , value in headers : <EOL> ikey = key . lower ( ) <EOL> if ikey == '<STR_LIT:location>' : <EOL> location = value <EOL> elif ikey == '<STR_LIT>' : <EOL> content_location = value <EOL> elif ikey == '<STR_LIT>' : <EOL> content_length = value <EOL> if location is not None : <EOL> old_location = location <EOL> if isinstance ( location , unicode ) : <EOL> location = iri_to_uri ( location ) <EOL> if self . autocorrect_location_header : <EOL> location = urlparse . urljoin ( <EOL> get_current_url ( environ , root_only = True ) , <EOL> location <EOL> ) <EOL> if location != old_location : <EOL> headers [ '<STR_LIT>' ] = location <EOL> if content_location is not None and isinstance ( content_location , unicode ) : <EOL> headers [ '<STR_LIT>' ] = iri_to_uri ( content_location ) <EOL> if <NUM_LIT:100> <= status < <NUM_LIT:200> or status == <NUM_LIT> : <EOL> headers [ '<STR_LIT>' ] = content_length = '<STR_LIT:0>' <EOL> elif status == <NUM_LIT> : <EOL> remove_entity_headers ( headers ) <EOL> if self . automatically_set_content_length and self . is_sequence and content_length is None and status != <NUM_LIT> : <EOL> try : <EOL> content_length = sum ( len ( str ( x ) ) for x in self . response ) <EOL> except UnicodeError : <EOL> pass <EOL> else : <EOL> headers [ '<STR_LIT>' ] = str ( content_length ) <EOL> return headers <EOL> def get_app_iter ( self , environ ) : <EOL> """<STR_LIT>""" <EOL> status = self . status_code <EOL> if environ [ '<STR_LIT>' ] == '<STR_LIT>' or <NUM_LIT:100> <= status < <NUM_LIT:200> or status in ( <NUM_LIT> , <NUM_LIT> ) : <EOL> return ( ) <EOL> if self . direct_passthrough : <EOL> if __debug__ : <EOL> _warn_if_string ( self . response ) <EOL> return self . response <EOL> return ClosingIterator ( self . iter_encoded ( ) , self . close ) <EOL> def get_wsgi_response ( self , environ ) : <EOL> """<STR_LIT>""" <EOL> if self . fix_headers . func_code is not BaseResponse . fix_headers . func_code : <EOL> if __debug__ : <EOL> from warnings import warn <EOL> warn ( DeprecationWarning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> stacklevel = <NUM_LIT:2> ) <EOL> self . fix_headers ( environ ) <EOL> headers = self . headers <EOL> else : <EOL> headers = self . get_wsgi_headers ( environ ) <EOL> app_iter = self . get_app_iter ( environ ) <EOL> return app_iter , self . status , headers . to_list ( ) <EOL> def __call__ ( self , environ , start_response ) : <EOL> """<STR_LIT>""" <EOL> app_iter , status , headers = self . get_wsgi_response ( environ ) <EOL> start_response ( status , headers ) <EOL> return app_iter <EOL> class AcceptMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def accept_mimetypes ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_accept_header ( self . environ . get ( '<STR_LIT>' ) , MIMEAccept ) <EOL> @ cached_property <EOL> def accept_charsets ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_accept_header ( self . environ . get ( '<STR_LIT>' ) , <EOL> CharsetAccept ) <EOL> @ cached_property <EOL> def accept_encodings ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_accept_header ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def accept_languages ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_accept_header ( self . environ . get ( '<STR_LIT>' ) , <EOL> LanguageAccept ) <EOL> class ETagRequestMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def cache_control ( self ) : <EOL> """<STR_LIT>""" <EOL> cache_control = self . environ . get ( '<STR_LIT>' ) <EOL> return parse_cache_control_header ( cache_control , None , <EOL> RequestCacheControl ) <EOL> @ cached_property <EOL> def if_match ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_etags ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def if_none_match ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_etags ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def if_modified_since ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_date ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def if_unmodified_since ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_date ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def if_range ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_if_range_header ( self . environ . get ( '<STR_LIT>' ) ) <EOL> @ cached_property <EOL> def range ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_range_header ( self . environ . get ( '<STR_LIT>' ) ) <EOL> class UserAgentMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def user_agent ( self ) : <EOL> """<STR_LIT>""" <EOL> from werkzeug . useragents import UserAgent <EOL> return UserAgent ( self . environ ) <EOL> class AuthorizationMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def authorization ( self ) : <EOL> """<STR_LIT>""" <EOL> header = self . environ . get ( '<STR_LIT>' ) <EOL> return parse_authorization_header ( header ) <EOL> class ETagResponseMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def cache_control ( self ) : <EOL> """<STR_LIT>""" <EOL> def on_update ( cache_control ) : <EOL> if not cache_control and '<STR_LIT>' in self . headers : <EOL> del self . headers [ '<STR_LIT>' ] <EOL> elif cache_control : <EOL> self . headers [ '<STR_LIT>' ] = cache_control . to_header ( ) <EOL> return parse_cache_control_header ( self . headers . get ( '<STR_LIT>' ) , <EOL> on_update , <EOL> ResponseCacheControl ) <EOL> def make_conditional ( self , request_or_environ ) : <EOL> """<STR_LIT>""" <EOL> environ = _get_environ ( request_or_environ ) <EOL> if environ [ '<STR_LIT>' ] in ( '<STR_LIT:GET>' , '<STR_LIT>' ) : <EOL> if '<STR_LIT:date>' not in self . headers : <EOL> self . headers [ '<STR_LIT>' ] = http_date ( ) <EOL> if '<STR_LIT>' not in self . headers : <EOL> self . headers [ '<STR_LIT>' ] = len ( self . data ) <EOL> if not is_resource_modified ( environ , self . headers . get ( '<STR_LIT>' ) , None , <EOL> self . headers . get ( '<STR_LIT>' ) ) : <EOL> self . status_code = <NUM_LIT> <EOL> return self <EOL> def add_etag ( self , overwrite = False , weak = False ) : <EOL> """<STR_LIT>""" <EOL> if overwrite or '<STR_LIT>' not in self . headers : <EOL> self . set_etag ( generate_etag ( self . data ) , weak ) <EOL> def set_etag ( self , etag , weak = False ) : <EOL> """<STR_LIT>""" <EOL> self . headers [ '<STR_LIT>' ] = quote_etag ( etag , weak ) <EOL> def get_etag ( self ) : <EOL> """<STR_LIT>""" <EOL> return unquote_etag ( self . headers . get ( '<STR_LIT>' ) ) <EOL> def freeze ( self , no_etag = False ) : <EOL> """<STR_LIT>""" <EOL> if not no_etag : <EOL> self . add_etag ( ) <EOL> super ( ETagResponseMixin , self ) . freeze ( ) <EOL> accept_ranges = header_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> def _get_content_range ( self ) : <EOL> def on_update ( rng ) : <EOL> if not rng : <EOL> del self . headers [ '<STR_LIT>' ] <EOL> else : <EOL> self . headers [ '<STR_LIT>' ] = rng . to_header ( ) <EOL> rv = parse_content_range_header ( self . headers . get ( '<STR_LIT>' ) , <EOL> on_update ) <EOL> if rv is None : <EOL> rv = ContentRange ( None , None , None , on_update = on_update ) <EOL> return rv <EOL> def _set_content_range ( self , value ) : <EOL> if not value : <EOL> del self . headers [ '<STR_LIT>' ] <EOL> elif isinstance ( value , basestring ) : <EOL> self . headers [ '<STR_LIT>' ] = value <EOL> else : <EOL> self . headers [ '<STR_LIT>' ] = value . to_header ( ) <EOL> content_range = property ( _get_content_range , _set_content_range , doc = '''<STR_LIT>''' ) <EOL> del _get_content_range , _set_content_range <EOL> class ResponseStream ( object ) : <EOL> """<STR_LIT>""" <EOL> mode = '<STR_LIT>' <EOL> def __init__ ( self , response ) : <EOL> self . response = response <EOL> self . closed = False <EOL> def write ( self , value ) : <EOL> if self . closed : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> self . response . _ensure_sequence ( mutable = True ) <EOL> self . response . response . append ( value ) <EOL> def writelines ( self , seq ) : <EOL> for item in seq : <EOL> self . write ( item ) <EOL> def close ( self ) : <EOL> self . closed = True <EOL> def flush ( self ) : <EOL> if self . closed : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> def isatty ( self ) : <EOL> if self . closed : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> return False <EOL> @ property <EOL> def encoding ( self ) : <EOL> return self . response . charset <EOL> class ResponseStreamMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ cached_property <EOL> def stream ( self ) : <EOL> """<STR_LIT>""" <EOL> return ResponseStream ( self ) <EOL> class CommonRequestDescriptorsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> content_type = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> content_length = environ_property ( '<STR_LIT>' , None , int , str , doc = '''<STR_LIT>''' ) <EOL> referrer = environ_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> date = environ_property ( '<STR_LIT>' , None , parse_date , doc = '''<STR_LIT>''' ) <EOL> max_forwards = environ_property ( '<STR_LIT>' , None , int , doc = '''<STR_LIT>''' ) <EOL> def _parse_content_type ( self ) : <EOL> if not hasattr ( self , '<STR_LIT>' ) : <EOL> self . _parsed_content_type = parse_options_header ( self . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> @ property <EOL> def mimetype ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _parse_content_type ( ) <EOL> return self . _parsed_content_type [ <NUM_LIT:0> ] <EOL> @ property <EOL> def mimetype_params ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _parse_content_type ( ) <EOL> return self . _parsed_content_type [ <NUM_LIT:1> ] <EOL> @ cached_property <EOL> def pragma ( self ) : <EOL> """<STR_LIT>""" <EOL> return parse_set_header ( self . environ . get ( '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> class CommonResponseDescriptorsMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> def _get_mimetype ( self ) : <EOL> ct = self . headers . get ( '<STR_LIT>' ) <EOL> if ct : <EOL> return ct . split ( '<STR_LIT:;>' ) [ <NUM_LIT:0> ] . strip ( ) <EOL> def _set_mimetype ( self , value ) : <EOL> self . headers [ '<STR_LIT:Content-Type>' ] = get_content_type ( value , self . charset ) <EOL> def _get_mimetype_params ( self ) : <EOL> def on_update ( d ) : <EOL> self . headers [ '<STR_LIT:Content-Type>' ] = dump_options_header ( self . mimetype , d ) <EOL> d = parse_options_header ( self . headers . get ( '<STR_LIT>' , '<STR_LIT>' ) ) [ <NUM_LIT:1> ] <EOL> return CallbackDict ( d , on_update ) <EOL> mimetype = property ( _get_mimetype , _set_mimetype , doc = '''<STR_LIT>''' ) <EOL> mimetype_params = property ( _get_mimetype_params , doc = '''<STR_LIT>''' ) <EOL> location = header_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> age = header_property ( '<STR_LIT>' , None , parse_date , http_date , doc = '''<STR_LIT>''' ) <EOL> content_type = header_property ( '<STR_LIT:Content-Type>' , doc = '''<STR_LIT>''' ) <EOL> content_length = header_property ( '<STR_LIT>' , None , int , str , doc = '''<STR_LIT>''' ) <EOL> content_location = header_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> content_encoding = header_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> content_md5 = header_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> date = header_property ( '<STR_LIT>' , None , parse_date , http_date , doc = '''<STR_LIT>''' ) <EOL> expires = header_property ( '<STR_LIT>' , None , parse_date , http_date , doc = '''<STR_LIT>''' ) <EOL> last_modified = header_property ( '<STR_LIT>' , None , parse_date , <EOL> http_date , doc = '''<STR_LIT>''' ) <EOL> def _get_retry_after ( self ) : <EOL> value = self . headers . get ( '<STR_LIT>' ) <EOL> if value is None : <EOL> return <EOL> elif value . isdigit ( ) : <EOL> return datetime . utcnow ( ) + timedelta ( seconds = int ( value ) ) <EOL> return parse_date ( value ) <EOL> def _set_retry_after ( self , value ) : <EOL> if value is None : <EOL> if '<STR_LIT>' in self . headers : <EOL> del self . headers [ '<STR_LIT>' ] <EOL> return <EOL> elif isinstance ( value , datetime ) : <EOL> value = http_date ( value ) <EOL> else : <EOL> value = str ( value ) <EOL> self . headers [ '<STR_LIT>' ] = value <EOL> retry_after = property ( _get_retry_after , _set_retry_after , doc = '''<STR_LIT>''' ) <EOL> def _set_property ( name , doc = None ) : <EOL> def fget ( self ) : <EOL> def on_update ( header_set ) : <EOL> if not header_set and name in self . headers : <EOL> del self . headers [ name ] <EOL> elif header_set : <EOL> self . headers [ name ] = header_set . to_header ( ) <EOL> return parse_set_header ( self . headers . get ( name ) , on_update ) <EOL> def fset ( self , value ) : <EOL> if not value : <EOL> del self . headers [ name ] <EOL> elif isinstance ( value , basestring ) : <EOL> self . headers [ name ] = value <EOL> else : <EOL> self . headers [ name ] = dump_header ( value ) <EOL> return property ( fget , fset , doc = doc ) <EOL> vary = _set_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> content_language = _set_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> allow = _set_property ( '<STR_LIT>' , doc = '''<STR_LIT>''' ) <EOL> del _set_property , _get_mimetype , _set_mimetype , _get_retry_after , _set_retry_after <EOL> class WWWAuthenticateMixin ( object ) : <EOL> """<STR_LIT>""" <EOL> @ property <EOL> def www_authenticate ( self ) : <EOL> """<STR_LIT>""" <EOL> def on_update ( www_auth ) : <EOL> if not www_auth and '<STR_LIT>' in self . headers : <EOL> del self . headers [ '<STR_LIT>' ] <EOL> elif www_auth : <EOL> self . headers [ '<STR_LIT>' ] = www_auth . to_header ( ) <EOL> header = self . headers . get ( '<STR_LIT>' ) <EOL> return parse_www_authenticate_header ( header , on_update ) <EOL> class Request ( BaseRequest , AcceptMixin , ETagRequestMixin , <EOL> UserAgentMixin , AuthorizationMixin , <EOL> CommonRequestDescriptorsMixin ) : <EOL> """<STR_LIT>""" <EOL> class Response ( BaseResponse , ETagResponseMixin , ResponseStreamMixin , <EOL> CommonResponseDescriptorsMixin , <EOL> WWWAuthenticateMixin ) : <EOL> """<STR_LIT>""" </s>
