<s> """<STR_LIT>""" <EOL> import collections <EOL> import datetime <EOL> from io import BytesIO , UnsupportedOperation <EOL> from . hooks import default_hooks <EOL> from . structures import CaseInsensitiveDict <EOL> from . auth import HTTPBasicAuth <EOL> from . cookies import cookiejar_from_dict , get_cookie_header , _copy_cookie_jar <EOL> from . packages . urllib3 . fields import RequestField <EOL> from . packages . urllib3 . filepost import encode_multipart_formdata <EOL> from . packages . urllib3 . util import parse_url <EOL> from . packages . urllib3 . exceptions import ( <EOL> DecodeError , ReadTimeoutError , ProtocolError , LocationParseError ) <EOL> from . exceptions import ( <EOL> HTTPError , MissingSchema , InvalidURL , ChunkedEncodingError , <EOL> ContentDecodingError , ConnectionError , StreamConsumedError ) <EOL> from . utils import ( <EOL> guess_filename , get_auth_from_url , requote_uri , <EOL> stream_decode_response_unicode , to_key_val_list , parse_header_links , <EOL> iter_slices , guess_json_utf , super_len , to_native_string ) <EOL> from . compat import ( <EOL> cookielib , urlunparse , urlsplit , urlencode , str , bytes , StringIO , <EOL> is_py2 , chardet , builtin_str , basestring ) <EOL> from . compat import json as complexjson <EOL> from . status_codes import codes <EOL> REDIRECT_STATI = ( <EOL> codes . moved , <EOL> codes . found , <EOL> codes . other , <EOL> codes . temporary_redirect , <EOL> codes . permanent_redirect , <EOL> ) <EOL> DEFAULT_REDIRECT_LIMIT = <NUM_LIT:30> <EOL> CONTENT_CHUNK_SIZE = <NUM_LIT:10> * <NUM_LIT> <EOL> ITER_CHUNK_SIZE = <NUM_LIT> <EOL> class RequestEncodingMixin ( object ) : <EOL> @ property <EOL> def path_url ( self ) : <EOL> """<STR_LIT>""" <EOL> url = [ ] <EOL> p = urlsplit ( self . url ) <EOL> path = p . path <EOL> if not path : <EOL> path = '<STR_LIT:/>' <EOL> url . append ( path ) <EOL> query = p . query <EOL> if query : <EOL> url . append ( '<STR_LIT:?>' ) <EOL> url . append ( query ) <EOL> return '<STR_LIT>' . join ( url ) <EOL> @ staticmethod <EOL> def _encode_params ( data ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , ( str , bytes ) ) : <EOL> return data <EOL> elif hasattr ( data , '<STR_LIT>' ) : <EOL> return data <EOL> elif hasattr ( data , '<STR_LIT>' ) : <EOL> result = [ ] <EOL> for k , vs in to_key_val_list ( data ) : <EOL> if isinstance ( vs , basestring ) or not hasattr ( vs , '<STR_LIT>' ) : <EOL> vs = [ vs ] <EOL> for v in vs : <EOL> if v is not None : <EOL> result . append ( <EOL> ( k . encode ( '<STR_LIT:utf-8>' ) if isinstance ( k , str ) else k , <EOL> v . encode ( '<STR_LIT:utf-8>' ) if isinstance ( v , str ) else v ) ) <EOL> return urlencode ( result , doseq = True ) <EOL> else : <EOL> return data <EOL> @ staticmethod <EOL> def _encode_files ( files , data ) : <EOL> """<STR_LIT>""" <EOL> if ( not files ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> elif isinstance ( data , basestring ) : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> new_fields = [ ] <EOL> fields = to_key_val_list ( data or { } ) <EOL> files = to_key_val_list ( files or { } ) <EOL> for field , val in fields : <EOL> if isinstance ( val , basestring ) or not hasattr ( val , '<STR_LIT>' ) : <EOL> val = [ val ] <EOL> for v in val : <EOL> if v is not None : <EOL> if not isinstance ( v , bytes ) : <EOL> v = str ( v ) <EOL> new_fields . append ( <EOL> ( field . decode ( '<STR_LIT:utf-8>' ) if isinstance ( field , bytes ) else field , <EOL> v . encode ( '<STR_LIT:utf-8>' ) if isinstance ( v , str ) else v ) ) <EOL> for ( k , v ) in files : <EOL> ft = None <EOL> fh = None <EOL> if isinstance ( v , ( tuple , list ) ) : <EOL> if len ( v ) == <NUM_LIT:2> : <EOL> fn , fp = v <EOL> elif len ( v ) == <NUM_LIT:3> : <EOL> fn , fp , ft = v <EOL> else : <EOL> fn , fp , ft , fh = v <EOL> else : <EOL> fn = guess_filename ( v ) or k <EOL> fp = v <EOL> if isinstance ( fp , ( str , bytes , bytearray ) ) : <EOL> fdata = fp <EOL> else : <EOL> fdata = fp . read ( ) <EOL> rf = RequestField ( name = k , data = fdata , filename = fn , headers = fh ) <EOL> rf . make_multipart ( content_type = ft ) <EOL> new_fields . append ( rf ) <EOL> body , content_type = encode_multipart_formdata ( new_fields ) <EOL> return body , content_type <EOL> class RequestHooksMixin ( object ) : <EOL> def register_hook ( self , event , hook ) : <EOL> """<STR_LIT>""" <EOL> if event not in self . hooks : <EOL> raise ValueError ( '<STR_LIT>' % ( event ) ) <EOL> if isinstance ( hook , collections . Callable ) : <EOL> self . hooks [ event ] . append ( hook ) <EOL> elif hasattr ( hook , '<STR_LIT>' ) : <EOL> self . hooks [ event ] . extend ( h for h in hook if isinstance ( h , collections . Callable ) ) <EOL> def deregister_hook ( self , event , hook ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . hooks [ event ] . remove ( hook ) <EOL> return True <EOL> except ValueError : <EOL> return False <EOL> class Request ( RequestHooksMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , method = None , url = None , headers = None , files = None , <EOL> data = None , params = None , auth = None , cookies = None , hooks = None , json = None ) : <EOL> data = [ ] if data is None else data <EOL> files = [ ] if files is None else files <EOL> headers = { } if headers is None else headers <EOL> params = { } if params is None else params <EOL> hooks = { } if hooks is None else hooks <EOL> self . hooks = default_hooks ( ) <EOL> for ( k , v ) in list ( hooks . items ( ) ) : <EOL> self . register_hook ( event = k , hook = v ) <EOL> self . method = method <EOL> self . url = url <EOL> self . headers = headers <EOL> self . files = files <EOL> self . data = data <EOL> self . json = json <EOL> self . params = params <EOL> self . auth = auth <EOL> self . cookies = cookies <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . method ) <EOL> def prepare ( self ) : <EOL> """<STR_LIT>""" <EOL> p = PreparedRequest ( ) <EOL> p . prepare ( <EOL> method = self . method , <EOL> url = self . url , <EOL> headers = self . headers , <EOL> files = self . files , <EOL> data = self . data , <EOL> json = self . json , <EOL> params = self . params , <EOL> auth = self . auth , <EOL> cookies = self . cookies , <EOL> hooks = self . hooks , <EOL> ) <EOL> return p <EOL> class PreparedRequest ( RequestEncodingMixin , RequestHooksMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . method = None <EOL> self . url = None <EOL> self . headers = None <EOL> self . _cookies = None <EOL> self . body = None <EOL> self . hooks = default_hooks ( ) <EOL> def prepare ( self , method = None , url = None , headers = None , files = None , <EOL> data = None , params = None , auth = None , cookies = None , hooks = None , json = None ) : <EOL> """<STR_LIT>""" <EOL> self . prepare_method ( method ) <EOL> self . prepare_url ( url , params ) <EOL> self . prepare_headers ( headers ) <EOL> self . prepare_cookies ( cookies ) <EOL> self . prepare_body ( data , files , json ) <EOL> self . prepare_auth ( auth , url ) <EOL> self . prepare_hooks ( hooks ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . method ) <EOL> def copy ( self ) : <EOL> p = PreparedRequest ( ) <EOL> p . method = self . method <EOL> p . url = self . url <EOL> p . headers = self . headers . copy ( ) if self . headers is not None else None <EOL> p . _cookies = _copy_cookie_jar ( self . _cookies ) <EOL> p . body = self . body <EOL> p . hooks = self . hooks <EOL> return p <EOL> def prepare_method ( self , method ) : <EOL> """<STR_LIT>""" <EOL> self . method = method <EOL> if self . method is not None : <EOL> self . method = to_native_string ( self . method . upper ( ) ) <EOL> def prepare_url ( self , url , params ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( url , bytes ) : <EOL> url = url . decode ( '<STR_LIT:utf8>' ) <EOL> else : <EOL> url = unicode ( url ) if is_py2 else str ( url ) <EOL> if '<STR_LIT::>' in url and not url . lower ( ) . startswith ( '<STR_LIT:http>' ) : <EOL> self . url = url <EOL> return <EOL> try : <EOL> scheme , auth , host , port , path , query , fragment = parse_url ( url ) <EOL> except LocationParseError as e : <EOL> raise InvalidURL ( * e . args ) <EOL> if not scheme : <EOL> error = ( "<STR_LIT>" ) <EOL> error = error . format ( to_native_string ( url , '<STR_LIT:utf8>' ) ) <EOL> raise MissingSchema ( error ) <EOL> if not host : <EOL> raise InvalidURL ( "<STR_LIT>" % url ) <EOL> try : <EOL> host = host . encode ( '<STR_LIT>' ) . decode ( '<STR_LIT:utf-8>' ) <EOL> except UnicodeError : <EOL> raise InvalidURL ( '<STR_LIT>' ) <EOL> netloc = auth or '<STR_LIT>' <EOL> if netloc : <EOL> netloc += '<STR_LIT:@>' <EOL> netloc += host <EOL> if port : <EOL> netloc += '<STR_LIT::>' + str ( port ) <EOL> if not path : <EOL> path = '<STR_LIT:/>' <EOL> if is_py2 : <EOL> if isinstance ( scheme , str ) : <EOL> scheme = scheme . encode ( '<STR_LIT:utf-8>' ) <EOL> if isinstance ( netloc , str ) : <EOL> netloc = netloc . encode ( '<STR_LIT:utf-8>' ) <EOL> if isinstance ( path , str ) : <EOL> path = path . encode ( '<STR_LIT:utf-8>' ) <EOL> if isinstance ( query , str ) : <EOL> query = query . encode ( '<STR_LIT:utf-8>' ) <EOL> if isinstance ( fragment , str ) : <EOL> fragment = fragment . encode ( '<STR_LIT:utf-8>' ) <EOL> if isinstance ( params , ( str , bytes ) ) : <EOL> params = to_native_string ( params ) <EOL> enc_params = self . _encode_params ( params ) <EOL> if enc_params : <EOL> if query : <EOL> query = '<STR_LIT>' % ( query , enc_params ) <EOL> else : <EOL> query = enc_params <EOL> url = requote_uri ( urlunparse ( [ scheme , netloc , path , None , query , fragment ] ) ) <EOL> self . url = url <EOL> def prepare_headers ( self , headers ) : <EOL> """<STR_LIT>""" <EOL> if headers : <EOL> self . headers = CaseInsensitiveDict ( ( to_native_string ( name ) , value ) for name , value in headers . items ( ) ) <EOL> else : <EOL> self . headers = CaseInsensitiveDict ( ) <EOL> def prepare_body ( self , data , files , json = None ) : <EOL> """<STR_LIT>""" <EOL> body = None <EOL> content_type = None <EOL> length = None <EOL> if not data and json is not None : <EOL> content_type = '<STR_LIT:application/json>' <EOL> body = complexjson . dumps ( json ) <EOL> is_stream = all ( [ <EOL> hasattr ( data , '<STR_LIT>' ) , <EOL> not isinstance ( data , ( basestring , list , tuple , dict ) ) <EOL> ] ) <EOL> try : <EOL> length = super_len ( data ) <EOL> except ( TypeError , AttributeError , UnsupportedOperation ) : <EOL> length = None <EOL> if is_stream : <EOL> body = data <EOL> if files : <EOL> raise NotImplementedError ( '<STR_LIT>' ) <EOL> if length : <EOL> self . headers [ '<STR_LIT>' ] = builtin_str ( length ) <EOL> else : <EOL> self . headers [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> if files : <EOL> ( body , content_type ) = self . _encode_files ( files , data ) <EOL> else : <EOL> if data : <EOL> body = self . _encode_params ( data ) <EOL> if isinstance ( data , basestring ) or hasattr ( data , '<STR_LIT>' ) : <EOL> content_type = None <EOL> else : <EOL> content_type = '<STR_LIT>' <EOL> self . prepare_content_length ( body ) <EOL> if content_type and ( '<STR_LIT>' not in self . headers ) : <EOL> self . headers [ '<STR_LIT:Content-Type>' ] = content_type <EOL> self . body = body <EOL> def prepare_content_length ( self , body ) : <EOL> if hasattr ( body , '<STR_LIT>' ) and hasattr ( body , '<STR_LIT>' ) : <EOL> curr_pos = body . tell ( ) <EOL> body . seek ( <NUM_LIT:0> , <NUM_LIT:2> ) <EOL> end_pos = body . tell ( ) <EOL> self . headers [ '<STR_LIT>' ] = builtin_str ( max ( <NUM_LIT:0> , end_pos - curr_pos ) ) <EOL> body . seek ( curr_pos , <NUM_LIT:0> ) <EOL> elif body is not None : <EOL> l = super_len ( body ) <EOL> if l : <EOL> self . headers [ '<STR_LIT>' ] = builtin_str ( l ) <EOL> elif ( self . method not in ( '<STR_LIT:GET>' , '<STR_LIT>' ) ) and ( self . headers . get ( '<STR_LIT>' ) is None ) : <EOL> self . headers [ '<STR_LIT>' ] = '<STR_LIT:0>' <EOL> def prepare_auth ( self , auth , url = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if auth is None : <EOL> url_auth = get_auth_from_url ( self . url ) <EOL> auth = url_auth if any ( url_auth ) else None <EOL> if auth : <EOL> if isinstance ( auth , tuple ) and len ( auth ) == <NUM_LIT:2> : <EOL> auth = HTTPBasicAuth ( * auth ) <EOL> r = auth ( self ) <EOL> self . __dict__ . update ( r . __dict__ ) <EOL> self . prepare_content_length ( self . body ) <EOL> def prepare_cookies ( self , cookies ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( cookies , cookielib . CookieJar ) : <EOL> self . _cookies = cookies <EOL> else : <EOL> self . _cookies = cookiejar_from_dict ( cookies ) <EOL> cookie_header = get_cookie_header ( self . _cookies , self ) <EOL> if cookie_header is not None : <EOL> self . headers [ '<STR_LIT>' ] = cookie_header <EOL> def prepare_hooks ( self , hooks ) : <EOL> """<STR_LIT>""" <EOL> hooks = hooks or [ ] <EOL> for event in hooks : <EOL> self . register_hook ( event , hooks [ event ] ) <EOL> class Response ( object ) : <EOL> """<STR_LIT>""" <EOL> __attrs__ = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:url>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> def __init__ ( self ) : <EOL> super ( Response , self ) . __init__ ( ) <EOL> self . _content = False <EOL> self . _content_consumed = False <EOL> self . status_code = None <EOL> self . headers = CaseInsensitiveDict ( ) <EOL> self . raw = None <EOL> self . url = None <EOL> self . encoding = None <EOL> self . history = [ ] <EOL> self . reason = None <EOL> self . cookies = cookiejar_from_dict ( { } ) <EOL> self . elapsed = datetime . timedelta ( <NUM_LIT:0> ) <EOL> self . request = None <EOL> def __getstate__ ( self ) : <EOL> if not self . _content_consumed : <EOL> self . content <EOL> return dict ( <EOL> ( attr , getattr ( self , attr , None ) ) <EOL> for attr in self . __attrs__ <EOL> ) <EOL> def __setstate__ ( self , state ) : <EOL> for name , value in state . items ( ) : <EOL> setattr ( self , name , value ) <EOL> setattr ( self , '<STR_LIT>' , True ) <EOL> setattr ( self , '<STR_LIT>' , None ) <EOL> def __repr__ ( self ) : <EOL> return '<STR_LIT>' % ( self . status_code ) <EOL> def __bool__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . ok <EOL> def __nonzero__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . ok <EOL> def __iter__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . iter_content ( <NUM_LIT> ) <EOL> @ property <EOL> def ok ( self ) : <EOL> try : <EOL> self . raise_for_status ( ) <EOL> except HTTPError : <EOL> return False <EOL> return True <EOL> @ property <EOL> def is_redirect ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT:location>' in self . headers and self . status_code in REDIRECT_STATI ) <EOL> @ property <EOL> def is_permanent_redirect ( self ) : <EOL> """<STR_LIT>""" <EOL> return ( '<STR_LIT:location>' in self . headers and self . status_code in ( codes . moved_permanently , codes . permanent_redirect ) ) <EOL> @ property <EOL> def apparent_encoding ( self ) : <EOL> """<STR_LIT>""" <EOL> return chardet . detect ( self . content ) [ '<STR_LIT>' ] <EOL> def iter_content ( self , chunk_size = <NUM_LIT:1> , decode_unicode = False ) : <EOL> """<STR_LIT>""" <EOL> def generate ( ) : <EOL> if hasattr ( self . raw , '<STR_LIT>' ) : <EOL> try : <EOL> for chunk in self . raw . stream ( chunk_size , decode_content = True ) : <EOL> yield chunk <EOL> except ProtocolError as e : <EOL> raise ChunkedEncodingError ( e ) <EOL> except DecodeError as e : <EOL> raise ContentDecodingError ( e ) <EOL> except ReadTimeoutError as e : <EOL> raise ConnectionError ( e ) <EOL> else : <EOL> while True : <EOL> chunk = self . raw . read ( chunk_size ) <EOL> if not chunk : <EOL> break <EOL> yield chunk <EOL> self . _content_consumed = True <EOL> if self . _content_consumed and isinstance ( self . _content , bool ) : <EOL> raise StreamConsumedError ( ) <EOL> reused_chunks = iter_slices ( self . _content , chunk_size ) <EOL> stream_chunks = generate ( ) <EOL> chunks = reused_chunks if self . _content_consumed else stream_chunks <EOL> if decode_unicode : <EOL> chunks = stream_decode_response_unicode ( chunks , self ) <EOL> return chunks <EOL> def iter_lines ( self , chunk_size = ITER_CHUNK_SIZE , decode_unicode = None , delimiter = None ) : <EOL> """<STR_LIT>""" <EOL> pending = None <EOL> for chunk in self . iter_content ( chunk_size = chunk_size , decode_unicode = decode_unicode ) : <EOL> if pending is not None : <EOL> chunk = pending + chunk <EOL> if delimiter : <EOL> lines = chunk . split ( delimiter ) <EOL> else : <EOL> lines = chunk . splitlines ( ) <EOL> if lines and lines [ - <NUM_LIT:1> ] and chunk and lines [ - <NUM_LIT:1> ] [ - <NUM_LIT:1> ] == chunk [ - <NUM_LIT:1> ] : <EOL> pending = lines . pop ( ) <EOL> else : <EOL> pending = None <EOL> for line in lines : <EOL> yield line <EOL> if pending is not None : <EOL> yield pending <EOL> @ property <EOL> def content ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . _content is False : <EOL> try : <EOL> if self . _content_consumed : <EOL> raise RuntimeError ( <EOL> '<STR_LIT>' ) <EOL> if self . status_code == <NUM_LIT:0> : <EOL> self . _content = None <EOL> else : <EOL> self . _content = bytes ( ) . join ( self . iter_content ( CONTENT_CHUNK_SIZE ) ) or bytes ( ) <EOL> except AttributeError : <EOL> self . _content = None <EOL> self . _content_consumed = True <EOL> return self . _content <EOL> @ property <EOL> def text ( self ) : <EOL> """<STR_LIT>""" <EOL> content = None <EOL> encoding = self . encoding <EOL> if not self . content : <EOL> return str ( '<STR_LIT>' ) <EOL> if self . encoding is None : <EOL> encoding = self . apparent_encoding <EOL> try : <EOL> content = str ( self . content , encoding , errors = '<STR_LIT:replace>' ) <EOL> except ( LookupError , TypeError ) : <EOL> content = str ( self . content , errors = '<STR_LIT:replace>' ) <EOL> return content <EOL> def json ( self , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not self . encoding and len ( self . content ) > <NUM_LIT:3> : <EOL> encoding = guess_json_utf ( self . content ) <EOL> if encoding is not None : <EOL> try : <EOL> return complexjson . loads ( <EOL> self . content . decode ( encoding ) , ** kwargs <EOL> ) <EOL> except UnicodeDecodeError : <EOL> pass <EOL> return complexjson . loads ( self . text , ** kwargs ) <EOL> @ property <EOL> def links ( self ) : <EOL> """<STR_LIT>""" <EOL> header = self . headers . get ( '<STR_LIT>' ) <EOL> l = { } <EOL> if header : <EOL> links = parse_header_links ( header ) <EOL> for link in links : <EOL> key = link . get ( '<STR_LIT>' ) or link . get ( '<STR_LIT:url>' ) <EOL> l [ key ] = link <EOL> return l <EOL> def raise_for_status ( self ) : <EOL> """<STR_LIT>""" <EOL> http_error_msg = '<STR_LIT>' <EOL> if <NUM_LIT> <= self . status_code < <NUM_LIT> : <EOL> http_error_msg = '<STR_LIT>' % ( self . status_code , self . reason , self . url ) <EOL> elif <NUM_LIT> <= self . status_code < <NUM_LIT> : <EOL> http_error_msg = '<STR_LIT>' % ( self . status_code , self . reason , self . url ) <EOL> if http_error_msg : <EOL> raise HTTPError ( http_error_msg , response = self ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . _content_consumed : <EOL> return self . raw . close ( ) <EOL> return self . raw . release_conn ( ) </s>
