<s> from datetime import datetime <EOL> try : <EOL> import socket <EOL> NativeStream = socket . _fileobject <EOL> except AttributeError : <EOL> import io <EOL> NativeStream = io . BufferedReader <EOL> import mimeparse <EOL> import six <EOL> from wsgiref . validate import InputWrapper <EOL> from falcon . errors import * <EOL> from falcon import util <EOL> from falcon . util . uri import parse_query_string , parse_host , unquote_string <EOL> from falcon import request_helpers as helpers <EOL> from six . moves import http_cookies <EOL> SimpleCookie = http_cookies . SimpleCookie <EOL> DEFAULT_ERROR_LOG_FORMAT = ( u'<STR_LIT>' <EOL> u'<STR_LIT>' ) <EOL> TRUE_STRINGS = ( '<STR_LIT:true>' , '<STR_LIT:True>' , '<STR_LIT:yes>' , '<STR_LIT:1>' ) <EOL> FALSE_STRINGS = ( '<STR_LIT:false>' , '<STR_LIT:False>' , '<STR_LIT>' , '<STR_LIT:0>' ) <EOL> WSGI_CONTENT_HEADERS = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> _maybe_wrap_wsgi_stream = True <EOL> strptime = datetime . strptime <EOL> now = datetime . now <EOL> class Request ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:path>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> context_type = None <EOL> def __init__ ( self , env , options = None ) : <EOL> global _maybe_wrap_wsgi_stream <EOL> self . env = env <EOL> self . options = options if options else RequestOptions ( ) <EOL> self . _wsgierrors = env [ '<STR_LIT>' ] <EOL> self . stream = env [ '<STR_LIT>' ] <EOL> self . method = env [ '<STR_LIT>' ] <EOL> path = env [ '<STR_LIT>' ] <EOL> if path : <EOL> if six . PY3 : <EOL> path = path . encode ( '<STR_LIT>' ) . decode ( '<STR_LIT:utf-8>' , '<STR_LIT:replace>' ) <EOL> if len ( path ) != <NUM_LIT:1> and path . endswith ( '<STR_LIT:/>' ) : <EOL> self . path = path [ : - <NUM_LIT:1> ] <EOL> else : <EOL> self . path = path <EOL> else : <EOL> self . path = '<STR_LIT:/>' <EOL> if '<STR_LIT>' in env : <EOL> self . query_string = env [ '<STR_LIT>' ] <EOL> if self . query_string : <EOL> self . _params = parse_query_string ( <EOL> self . query_string , <EOL> keep_blank_qs_values = self . options . keep_blank_qs_values , <EOL> ) <EOL> else : <EOL> self . _params = { } <EOL> else : <EOL> self . query_string = '<STR_LIT>' <EOL> self . _params = { } <EOL> self . _cookies = None <EOL> self . _cached_headers = None <EOL> self . _cached_uri = None <EOL> self . _cached_relative_uri = None <EOL> self . _cached_access_route = None <EOL> try : <EOL> self . content_type = self . env [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> self . content_type = None <EOL> if _maybe_wrap_wsgi_stream : <EOL> if isinstance ( self . stream , ( NativeStream , InputWrapper , ) ) : <EOL> self . _wrap_stream ( ) <EOL> else : <EOL> _maybe_wrap_wsgi_stream = False <EOL> if ( self . options . auto_parse_form_urlencoded and <EOL> self . content_type is not None and <EOL> '<STR_LIT>' in self . content_type ) : <EOL> self . _parse_form_urlencoded ( ) <EOL> if self . context_type is None : <EOL> self . context = { } <EOL> else : <EOL> self . context = self . context_type ( ) <EOL> user_agent = helpers . header_property ( '<STR_LIT>' ) <EOL> auth = helpers . header_property ( '<STR_LIT>' ) <EOL> expect = helpers . header_property ( '<STR_LIT>' ) <EOL> if_match = helpers . header_property ( '<STR_LIT>' ) <EOL> if_none_match = helpers . header_property ( '<STR_LIT>' ) <EOL> if_range = helpers . header_property ( '<STR_LIT>' ) <EOL> @ property <EOL> def client_accepts_json ( self ) : <EOL> return self . client_accepts ( '<STR_LIT:application/json>' ) <EOL> @ property <EOL> def client_accepts_msgpack ( self ) : <EOL> return ( self . client_accepts ( '<STR_LIT>' ) or <EOL> self . client_accepts ( '<STR_LIT>' ) ) <EOL> @ property <EOL> def client_accepts_xml ( self ) : <EOL> return self . client_accepts ( '<STR_LIT>' ) <EOL> @ property <EOL> def accept ( self ) : <EOL> try : <EOL> return self . env [ '<STR_LIT>' ] or '<STR_LIT>' <EOL> except KeyError : <EOL> return '<STR_LIT>' <EOL> @ property <EOL> def content_length ( self ) : <EOL> try : <EOL> value = self . env [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> return None <EOL> if not value : <EOL> return None <EOL> try : <EOL> value_as_int = int ( value ) <EOL> except ValueError : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> if value_as_int < <NUM_LIT:0> : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> return value_as_int <EOL> @ property <EOL> def date ( self ) : <EOL> return self . get_header_as_datetime ( '<STR_LIT>' ) <EOL> @ property <EOL> def if_modified_since ( self ) : <EOL> return self . get_header_as_datetime ( '<STR_LIT>' ) <EOL> @ property <EOL> def if_unmodified_since ( self ) : <EOL> return self . get_header_as_datetime ( '<STR_LIT>' ) <EOL> @ property <EOL> def range ( self ) : <EOL> try : <EOL> value = self . env [ '<STR_LIT>' ] <EOL> if '<STR_LIT:=>' in value : <EOL> unit , sep , req_range = value . partition ( '<STR_LIT:=>' ) <EOL> else : <EOL> msg = "<STR_LIT>" <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> except KeyError : <EOL> return None <EOL> if '<STR_LIT:U+002C>' in req_range : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> try : <EOL> first , sep , last = req_range . partition ( '<STR_LIT:->' ) <EOL> if not sep : <EOL> raise ValueError ( ) <EOL> if first : <EOL> return ( int ( first ) , int ( last or - <NUM_LIT:1> ) ) <EOL> elif last : <EOL> return ( - int ( last ) , - <NUM_LIT:1> ) <EOL> else : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> except ValueError : <EOL> href = '<STR_LIT>' <EOL> href_text = '<STR_LIT>' <EOL> msg = ( '<STR_LIT>' ) <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' , href = href , <EOL> href_text = href_text ) <EOL> @ property <EOL> def range_unit ( self ) : <EOL> try : <EOL> value = self . env [ '<STR_LIT>' ] <EOL> if '<STR_LIT:=>' in value : <EOL> unit , sep , req_range = value . partition ( '<STR_LIT:=>' ) <EOL> return unit <EOL> else : <EOL> msg = "<STR_LIT>" <EOL> raise HTTPInvalidHeader ( msg , '<STR_LIT>' ) <EOL> except KeyError : <EOL> return None <EOL> @ property <EOL> def app ( self ) : <EOL> return self . env . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> @ property <EOL> def protocol ( self ) : <EOL> return self . env [ '<STR_LIT>' ] <EOL> @ property <EOL> def uri ( self ) : <EOL> if self . _cached_uri is None : <EOL> env = self . env <EOL> protocol = env [ '<STR_LIT>' ] <EOL> try : <EOL> host = env [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> host = env [ '<STR_LIT>' ] <EOL> port = env [ '<STR_LIT>' ] <EOL> if protocol == '<STR_LIT>' : <EOL> if port != '<STR_LIT>' : <EOL> host += '<STR_LIT::>' + port <EOL> else : <EOL> if port != '<STR_LIT>' : <EOL> host += '<STR_LIT::>' + port <EOL> value = ( protocol + '<STR_LIT>' + <EOL> host + <EOL> self . app + <EOL> self . path ) <EOL> if self . query_string : <EOL> value = value + '<STR_LIT:?>' + self . query_string <EOL> self . _cached_uri = value <EOL> return self . _cached_uri <EOL> url = uri <EOL> @ property <EOL> def host ( self ) : <EOL> try : <EOL> host_header = self . env [ '<STR_LIT>' ] <EOL> host , port = parse_host ( host_header ) <EOL> except KeyError : <EOL> host = self . env [ '<STR_LIT>' ] <EOL> return host <EOL> @ property <EOL> def subdomain ( self ) : <EOL> subdomain , sep , remainder = self . host . partition ( '<STR_LIT:.>' ) <EOL> return subdomain if sep else None <EOL> @ property <EOL> def relative_uri ( self ) : <EOL> if self . _cached_relative_uri is None : <EOL> if self . query_string : <EOL> self . _cached_relative_uri = ( self . app + self . path + '<STR_LIT:?>' + <EOL> self . query_string ) <EOL> else : <EOL> self . _cached_relative_uri = self . app + self . path <EOL> return self . _cached_relative_uri <EOL> @ property <EOL> def headers ( self ) : <EOL> if self . _cached_headers is None : <EOL> headers = self . _cached_headers = { } <EOL> env = self . env <EOL> for name , value in env . items ( ) : <EOL> if name . startswith ( '<STR_LIT>' ) : <EOL> headers [ name [ <NUM_LIT:5> : ] . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) ] = value <EOL> elif name in WSGI_CONTENT_HEADERS : <EOL> headers [ name . replace ( '<STR_LIT:_>' , '<STR_LIT:->' ) ] = value <EOL> return self . _cached_headers . copy ( ) <EOL> @ property <EOL> def params ( self ) : <EOL> return self . _params <EOL> @ property <EOL> def cookies ( self ) : <EOL> if self . _cookies is None : <EOL> parser = SimpleCookie ( self . get_header ( '<STR_LIT>' ) ) <EOL> cookies = { } <EOL> for morsel in parser . values ( ) : <EOL> cookies [ morsel . key ] = morsel . value <EOL> self . _cookies = cookies <EOL> return self . _cookies . copy ( ) <EOL> @ property <EOL> def access_route ( self ) : <EOL> if self . _cached_access_route is None : <EOL> if '<STR_LIT>' in self . env : <EOL> self . _cached_access_route = self . _parse_rfc_forwarded ( ) <EOL> elif '<STR_LIT>' in self . env : <EOL> addresses = self . env [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) <EOL> self . _cached_access_route = [ ip . strip ( ) for ip in addresses ] <EOL> elif '<STR_LIT>' in self . env : <EOL> self . _cached_access_route = [ self . env [ '<STR_LIT>' ] ] <EOL> elif '<STR_LIT>' in self . env : <EOL> self . _cached_access_route = [ self . env [ '<STR_LIT>' ] ] <EOL> else : <EOL> self . _cached_access_route = [ ] <EOL> return self . _cached_access_route <EOL> @ property <EOL> def remote_addr ( self ) : <EOL> return self . env . get ( '<STR_LIT>' ) <EOL> def client_accepts ( self , media_type ) : <EOL> """<STR_LIT>""" <EOL> accept = self . accept <EOL> if ( accept == media_type ) or ( accept == '<STR_LIT>' ) : <EOL> return True <EOL> try : <EOL> return mimeparse . quality ( media_type , accept ) != <NUM_LIT:0.0> <EOL> except ValueError : <EOL> return False <EOL> def client_prefers ( self , media_types ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> preferred_type = mimeparse . best_match ( media_types , self . accept ) <EOL> except ValueError : <EOL> preferred_type = '<STR_LIT>' <EOL> return ( preferred_type if preferred_type else None ) <EOL> def get_header ( self , name , required = False ) : <EOL> """<STR_LIT>""" <EOL> wsgi_name = name . upper ( ) . replace ( '<STR_LIT:->' , '<STR_LIT:_>' ) <EOL> try : <EOL> return self . env [ '<STR_LIT>' + wsgi_name ] <EOL> except KeyError : <EOL> if wsgi_name in WSGI_CONTENT_HEADERS : <EOL> try : <EOL> return self . env [ wsgi_name ] <EOL> except KeyError : <EOL> pass <EOL> if not required : <EOL> return None <EOL> raise HTTPMissingHeader ( name ) <EOL> def get_header_as_datetime ( self , header , required = False , obs_date = False ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> http_date = self . get_header ( header , required = required ) <EOL> return util . http_date_to_dt ( http_date , obs_date = obs_date ) <EOL> except TypeError : <EOL> return None <EOL> except ValueError : <EOL> msg = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> raise HTTPInvalidHeader ( msg , header ) <EOL> def get_param ( self , name , required = False , store = None , default = None ) : <EOL> """<STR_LIT>""" <EOL> params = self . _params <EOL> if name in params : <EOL> param = params [ name ] <EOL> if isinstance ( param , list ) : <EOL> param = param [ - <NUM_LIT:1> ] <EOL> if store is not None : <EOL> store [ name ] = param <EOL> return param <EOL> if not required : <EOL> return default <EOL> raise HTTPMissingParam ( name ) <EOL> def get_param_as_int ( self , name , <EOL> required = False , min = None , max = None , store = None ) : <EOL> """<STR_LIT>""" <EOL> params = self . _params <EOL> if name in params : <EOL> val = params [ name ] <EOL> if isinstance ( val , list ) : <EOL> val = val [ - <NUM_LIT:1> ] <EOL> try : <EOL> val = int ( val ) <EOL> except ValueError : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if min is not None and val < min : <EOL> msg = '<STR_LIT>' + str ( min ) <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if max is not None and max < val : <EOL> msg = '<STR_LIT>' + str ( max ) <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if store is not None : <EOL> store [ name ] = val <EOL> return val <EOL> if not required : <EOL> return None <EOL> raise HTTPMissingParam ( name ) <EOL> def get_param_as_bool ( self , name , required = False , store = None , <EOL> blank_as_true = False ) : <EOL> """<STR_LIT>""" <EOL> params = self . _params <EOL> if name in params : <EOL> val = params [ name ] <EOL> if isinstance ( val , list ) : <EOL> val = val [ - <NUM_LIT:1> ] <EOL> if val in TRUE_STRINGS : <EOL> val = True <EOL> elif val in FALSE_STRINGS : <EOL> val = False <EOL> elif blank_as_true and not val : <EOL> val = True <EOL> else : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if store is not None : <EOL> store [ name ] = val <EOL> return val <EOL> if not required : <EOL> return None <EOL> raise HTTPMissingParam ( name ) <EOL> def get_param_as_list ( self , name , <EOL> transform = None , required = False , store = None ) : <EOL> """<STR_LIT>""" <EOL> params = self . _params <EOL> if name in params : <EOL> items = params [ name ] <EOL> if not isinstance ( items , list ) : <EOL> items = [ items ] <EOL> if transform is not None : <EOL> try : <EOL> items = [ transform ( i ) for i in items ] <EOL> except ValueError : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if store is not None : <EOL> store [ name ] = items <EOL> return items <EOL> if not required : <EOL> return None <EOL> raise HTTPMissingParam ( name ) <EOL> def get_param_as_date ( self , name , format_string = '<STR_LIT>' , <EOL> required = False , store = None ) : <EOL> """<STR_LIT>""" <EOL> param_value = self . get_param ( name , required = required ) <EOL> if param_value is None : <EOL> return None <EOL> try : <EOL> date = strptime ( param_value , format_string ) . date ( ) <EOL> except ValueError : <EOL> msg = '<STR_LIT>' <EOL> raise HTTPInvalidParam ( msg , name ) <EOL> if store is not None : <EOL> store [ name ] = date <EOL> return date <EOL> def log_error ( self , message ) : <EOL> """<STR_LIT>""" <EOL> if self . query_string : <EOL> query_string_formatted = '<STR_LIT:?>' + self . query_string <EOL> else : <EOL> query_string_formatted = '<STR_LIT>' <EOL> log_line = ( <EOL> DEFAULT_ERROR_LOG_FORMAT . <EOL> format ( now ( ) , self . method , self . path , query_string_formatted ) <EOL> ) <EOL> if six . PY3 : <EOL> self . _wsgierrors . write ( log_line + message + '<STR_LIT:\n>' ) <EOL> else : <EOL> if isinstance ( message , unicode ) : <EOL> message = message . encode ( '<STR_LIT:utf-8>' ) <EOL> self . _wsgierrors . write ( log_line . encode ( '<STR_LIT:utf-8>' ) ) <EOL> self . _wsgierrors . write ( message + '<STR_LIT:\n>' ) <EOL> def _wrap_stream ( self ) : <EOL> try : <EOL> content_length = self . content_length or <NUM_LIT:0> <EOL> except HTTPInvalidHeader : <EOL> content_length = <NUM_LIT:0> <EOL> self . stream = helpers . Body ( self . stream , content_length ) <EOL> def _parse_form_urlencoded ( self ) : <EOL> body = self . stream . read ( ) <EOL> try : <EOL> body = body . decode ( '<STR_LIT:ascii>' ) <EOL> except UnicodeDecodeError : <EOL> body = None <EOL> self . log_error ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if body : <EOL> extra_params = parse_query_string ( <EOL> body , <EOL> keep_blank_qs_values = self . options . keep_blank_qs_values , <EOL> ) <EOL> self . _params . update ( extra_params ) <EOL> def _parse_rfc_forwarded ( self ) : <EOL> """<STR_LIT>""" <EOL> addr = [ ] <EOL> for forwarded in self . env [ '<STR_LIT>' ] . split ( '<STR_LIT:U+002C>' ) : <EOL> for param in forwarded . split ( '<STR_LIT:;>' ) : <EOL> key , _ , val = param . strip ( ) . partition ( '<STR_LIT:=>' ) <EOL> if not val : <EOL> continue <EOL> if key . lower ( ) != '<STR_LIT>' : <EOL> continue <EOL> host , _ = parse_host ( unquote_string ( val ) ) <EOL> addr . append ( host ) <EOL> return addr <EOL> class RequestOptions ( object ) : <EOL> """<STR_LIT>""" <EOL> __slots__ = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ) <EOL> def __init__ ( self ) : <EOL> self . keep_blank_qs_values = False <EOL> self . auto_parse_form_urlencoded = False </s>
