<s> """<STR_LIT>""" <EOL> import os <EOL> from collections import Mapping <EOL> from datetime import datetime <EOL> from . compat import cookielib , OrderedDict , urljoin , urlparse , builtin_str <EOL> from . cookies import ( <EOL> cookiejar_from_dict , extract_cookies_to_jar , RequestsCookieJar , merge_cookies ) <EOL> from . models import Request , PreparedRequest <EOL> from . hooks import default_hooks , dispatch_hook <EOL> from . utils import to_key_val_list , default_headers <EOL> from . exceptions import TooManyRedirects , InvalidSchema <EOL> from . structures import CaseInsensitiveDict <EOL> from . adapters import HTTPAdapter <EOL> from . utils import requote_uri , get_environ_proxies , get_netrc_auth <EOL> from . status_codes import codes <EOL> REDIRECT_STATI = ( <EOL> codes . moved , <EOL> codes . found , <EOL> codes . other , <EOL> codes . temporary_moved , <EOL> ) <EOL> DEFAULT_REDIRECT_LIMIT = <NUM_LIT:30> <EOL> def merge_setting ( request_setting , session_setting , dict_class = OrderedDict ) : <EOL> """<STR_LIT>""" <EOL> if session_setting is None : <EOL> return request_setting <EOL> if request_setting is None : <EOL> return session_setting <EOL> if not ( <EOL> isinstance ( session_setting , Mapping ) and <EOL> isinstance ( request_setting , Mapping ) <EOL> ) : <EOL> return request_setting <EOL> merged_setting = dict_class ( to_key_val_list ( session_setting ) ) <EOL> merged_setting . update ( to_key_val_list ( request_setting ) ) <EOL> for ( k , v ) in request_setting . items ( ) : <EOL> if v is None : <EOL> del merged_setting [ k ] <EOL> return merged_setting <EOL> def merge_hooks ( request_hooks , session_hooks , dict_class = OrderedDict ) : <EOL> """<STR_LIT>""" <EOL> if session_hooks is None or session_hooks . get ( '<STR_LIT>' ) == [ ] : <EOL> return request_hooks <EOL> if request_hooks is None or request_hooks . get ( '<STR_LIT>' ) == [ ] : <EOL> return session_hooks <EOL> return merge_setting ( request_hooks , session_hooks , dict_class ) <EOL> class SessionRedirectMixin ( object ) : <EOL> def resolve_redirects ( self , resp , req , stream = False , timeout = None , <EOL> verify = True , cert = None , proxies = None ) : <EOL> """<STR_LIT>""" <EOL> i = <NUM_LIT:0> <EOL> while ( '<STR_LIT:location>' in resp . headers and resp . status_code in REDIRECT_STATI ) : <EOL> prepared_request = req . copy ( ) <EOL> resp . content <EOL> if i >= self . max_redirects : <EOL> raise TooManyRedirects ( '<STR_LIT>' % self . max_redirects ) <EOL> resp . close ( ) <EOL> url = resp . headers [ '<STR_LIT:location>' ] <EOL> method = req . method <EOL> if url . startswith ( '<STR_LIT>' ) : <EOL> parsed_rurl = urlparse ( resp . url ) <EOL> url = '<STR_LIT>' % ( parsed_rurl . scheme , url ) <EOL> parsed = urlparse ( url ) <EOL> url = parsed . geturl ( ) <EOL> if not urlparse ( url ) . netloc : <EOL> url = urljoin ( resp . url , requote_uri ( url ) ) <EOL> else : <EOL> url = requote_uri ( url ) <EOL> prepared_request . url = url <EOL> if ( resp . status_code == codes . see_other and <EOL> method != '<STR_LIT>' ) : <EOL> method = '<STR_LIT:GET>' <EOL> if resp . status_code == codes . found and method != '<STR_LIT>' : <EOL> method = '<STR_LIT:GET>' <EOL> if resp . status_code == codes . moved and method == '<STR_LIT:POST>' : <EOL> method = '<STR_LIT:GET>' <EOL> prepared_request . method = method <EOL> if resp . status_code not in ( codes . temporary , codes . resume ) : <EOL> if '<STR_LIT>' in prepared_request . headers : <EOL> del prepared_request . headers [ '<STR_LIT>' ] <EOL> prepared_request . body = None <EOL> headers = prepared_request . headers <EOL> try : <EOL> del headers [ '<STR_LIT>' ] <EOL> except KeyError : <EOL> pass <EOL> extract_cookies_to_jar ( prepared_request . _cookies , <EOL> prepared_request , resp . raw ) <EOL> prepared_request . _cookies . update ( self . cookies ) <EOL> prepared_request . prepare_cookies ( prepared_request . _cookies ) <EOL> resp = self . send ( <EOL> prepared_request , <EOL> stream = stream , <EOL> timeout = timeout , <EOL> verify = verify , <EOL> cert = cert , <EOL> proxies = proxies , <EOL> allow_redirects = False , <EOL> ) <EOL> extract_cookies_to_jar ( self . cookies , prepared_request , resp . raw ) <EOL> i += <NUM_LIT:1> <EOL> yield resp <EOL> class Session ( SessionRedirectMixin ) : <EOL> """<STR_LIT>""" <EOL> __attrs__ = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> def __init__ ( self ) : <EOL> self . headers = default_headers ( ) <EOL> self . auth = None <EOL> self . proxies = { } <EOL> self . hooks = default_hooks ( ) <EOL> self . params = { } <EOL> self . stream = False <EOL> self . verify = True <EOL> self . cert = None <EOL> self . max_redirects = DEFAULT_REDIRECT_LIMIT <EOL> self . trust_env = True <EOL> self . cookies = cookiejar_from_dict ( { } ) <EOL> self . adapters = OrderedDict ( ) <EOL> self . mount ( '<STR_LIT>' , HTTPAdapter ( ) ) <EOL> self . mount ( '<STR_LIT>' , HTTPAdapter ( ) ) <EOL> def __enter__ ( self ) : <EOL> return self <EOL> def __exit__ ( self , * args ) : <EOL> self . close ( ) <EOL> def prepare_request ( self , request ) : <EOL> """<STR_LIT>""" <EOL> cookies = request . cookies or { } <EOL> if not isinstance ( cookies , cookielib . CookieJar ) : <EOL> cookies = cookiejar_from_dict ( cookies ) <EOL> merged_cookies = merge_cookies ( <EOL> merge_cookies ( RequestsCookieJar ( ) , self . cookies ) , cookies ) <EOL> auth = request . auth <EOL> if self . trust_env and not auth and not self . auth : <EOL> auth = get_netrc_auth ( request . url ) <EOL> p = PreparedRequest ( ) <EOL> p . prepare ( <EOL> method = request . method . upper ( ) , <EOL> url = request . url , <EOL> files = request . files , <EOL> data = request . data , <EOL> headers = merge_setting ( request . headers , self . headers , dict_class = CaseInsensitiveDict ) , <EOL> params = merge_setting ( request . params , self . params ) , <EOL> auth = merge_setting ( auth , self . auth ) , <EOL> cookies = merged_cookies , <EOL> hooks = merge_hooks ( request . hooks , self . hooks ) , <EOL> ) <EOL> return p <EOL> def request ( self , method , url , <EOL> params = None , <EOL> data = None , <EOL> headers = None , <EOL> cookies = None , <EOL> files = None , <EOL> auth = None , <EOL> timeout = None , <EOL> allow_redirects = True , <EOL> proxies = None , <EOL> hooks = None , <EOL> stream = None , <EOL> verify = None , <EOL> cert = None ) : <EOL> """<STR_LIT>""" <EOL> method = builtin_str ( method ) <EOL> req = Request ( <EOL> method = method . upper ( ) , <EOL> url = url , <EOL> headers = headers , <EOL> files = files , <EOL> data = data or { } , <EOL> params = params or { } , <EOL> auth = auth , <EOL> cookies = cookies , <EOL> hooks = hooks , <EOL> ) <EOL> prep = self . prepare_request ( req ) <EOL> proxies = proxies or { } <EOL> if self . trust_env : <EOL> env_proxies = get_environ_proxies ( url ) or { } <EOL> for ( k , v ) in env_proxies . items ( ) : <EOL> proxies . setdefault ( k , v ) <EOL> if not verify and verify is not False : <EOL> verify = os . environ . get ( '<STR_LIT>' ) <EOL> if not verify and verify is not False : <EOL> verify = os . environ . get ( '<STR_LIT>' ) <EOL> proxies = merge_setting ( proxies , self . proxies ) <EOL> stream = merge_setting ( stream , self . stream ) <EOL> verify = merge_setting ( verify , self . verify ) <EOL> cert = merge_setting ( cert , self . cert ) <EOL> send_kwargs = { <EOL> '<STR_LIT>' : stream , <EOL> '<STR_LIT>' : timeout , <EOL> '<STR_LIT>' : verify , <EOL> '<STR_LIT>' : cert , <EOL> '<STR_LIT>' : proxies , <EOL> '<STR_LIT>' : allow_redirects , <EOL> } <EOL> resp = self . send ( prep , ** send_kwargs ) <EOL> return resp <EOL> def get ( self , url , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> return self . request ( '<STR_LIT:GET>' , url , ** kwargs ) <EOL> def options ( self , url , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , True ) <EOL> return self . request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def head ( self , url , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , False ) <EOL> return self . request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def post ( self , url , data = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . request ( '<STR_LIT:POST>' , url , data = data , ** kwargs ) <EOL> def put ( self , url , data = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . request ( '<STR_LIT>' , url , data = data , ** kwargs ) <EOL> def patch ( self , url , data = None , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . request ( '<STR_LIT>' , url , data = data , ** kwargs ) <EOL> def delete ( self , url , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> return self . request ( '<STR_LIT>' , url , ** kwargs ) <EOL> def send ( self , request , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> kwargs . setdefault ( '<STR_LIT>' , self . stream ) <EOL> kwargs . setdefault ( '<STR_LIT>' , self . verify ) <EOL> kwargs . setdefault ( '<STR_LIT>' , self . cert ) <EOL> kwargs . setdefault ( '<STR_LIT>' , self . proxies ) <EOL> if not isinstance ( request , PreparedRequest ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> allow_redirects = kwargs . pop ( '<STR_LIT>' , True ) <EOL> stream = kwargs . get ( '<STR_LIT>' ) <EOL> timeout = kwargs . get ( '<STR_LIT>' ) <EOL> verify = kwargs . get ( '<STR_LIT>' ) <EOL> cert = kwargs . get ( '<STR_LIT>' ) <EOL> proxies = kwargs . get ( '<STR_LIT>' ) <EOL> hooks = request . hooks <EOL> adapter = self . get_adapter ( url = request . url ) <EOL> start = datetime . utcnow ( ) <EOL> r = adapter . send ( request , ** kwargs ) <EOL> r . elapsed = datetime . utcnow ( ) - start <EOL> r = dispatch_hook ( '<STR_LIT>' , hooks , r , ** kwargs ) <EOL> if r . history : <EOL> for resp in r . history : <EOL> extract_cookies_to_jar ( self . cookies , resp . request , resp . raw ) <EOL> extract_cookies_to_jar ( self . cookies , request , r . raw ) <EOL> gen = self . resolve_redirects ( r , request , stream = stream , <EOL> timeout = timeout , verify = verify , cert = cert , <EOL> proxies = proxies ) <EOL> history = [ resp for resp in gen ] if allow_redirects else [ ] <EOL> if history : <EOL> history . insert ( <NUM_LIT:0> , r ) <EOL> r = history . pop ( ) <EOL> r . history = tuple ( history ) <EOL> return r <EOL> def get_adapter ( self , url ) : <EOL> """<STR_LIT>""" <EOL> for ( prefix , adapter ) in self . adapters . items ( ) : <EOL> if url . lower ( ) . startswith ( prefix ) : <EOL> return adapter <EOL> raise InvalidSchema ( "<STR_LIT>" % url ) <EOL> def close ( self ) : <EOL> """<STR_LIT>""" <EOL> for v in self . adapters . values ( ) : <EOL> v . close ( ) <EOL> def mount ( self , prefix , adapter ) : <EOL> """<STR_LIT>""" <EOL> self . adapters [ prefix ] = adapter <EOL> keys_to_move = [ k for k in self . adapters if len ( k ) < len ( prefix ) ] <EOL> for key in keys_to_move : <EOL> self . adapters [ key ] = self . adapters . pop ( key ) <EOL> def __getstate__ ( self ) : <EOL> return dict ( ( attr , getattr ( self , attr , None ) ) for attr in self . __attrs__ ) <EOL> def __setstate__ ( self , state ) : <EOL> for attr , value in state . items ( ) : <EOL> setattr ( self , attr , value ) <EOL> def session ( ) : <EOL> """<STR_LIT>""" <EOL> return Session ( ) </s>
