<s> """<STR_LIT>""" <EOL> import re <EOL> from glob import glob <EOL> from six import iteritems <EOL> from six . moves . urllib . parse import urlparse <EOL> from os . path import dirname , abspath , join as pathjoin <EOL> from . . dochelpers import exc_str <EOL> from . . support . configparserinc import SafeConfigParserWithIncludes <EOL> from . . ui import ui <EOL> from . . utils import auto_repr <EOL> from . . utils import assure_list_from_str <EOL> from . base import NoneAuthenticator , NotImplementedAuthenticator <EOL> from . http import HTMLFormAuthenticator , HTTPBasicAuthAuthenticator , HTTPDigestAuthAuthenticator <EOL> from . s3 import S3Authenticator , S3Downloader <EOL> from logging import getLogger <EOL> lgr = getLogger ( '<STR_LIT>' ) <EOL> def resolve_url_to_name ( d , url ) : <EOL> """<STR_LIT>""" <EOL> for k , rec in iteritems ( d ) : <EOL> for url_re in rec . get ( '<STR_LIT>' , '<STR_LIT>' ) . split ( '<STR_LIT:\n>' ) : <EOL> if url_re : <EOL> if re . search ( url_re , url ) : <EOL> return k <EOL> return None <EOL> from . . support . keyring_ import keyring <EOL> @ auto_repr <EOL> class Credential ( object ) : <EOL> TYPES = { <EOL> '<STR_LIT>' : ( '<STR_LIT:user>' , '<STR_LIT:password>' ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> } <EOL> def __init__ ( self , name , type , url ) : <EOL> self . name = name <EOL> if not type in self . TYPES : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % ( type , self . TYPES . keys ( ) ) ) <EOL> self . type = type <EOL> self . url = url <EOL> def _ask_field_value ( self , f ) : <EOL> return ui . question ( f , <EOL> title = "<STR_LIT>" % self . name <EOL> + "<STR_LIT>" <EOL> % self . url if self . url else '<STR_LIT>' , <EOL> hidden = True ) <EOL> def enter_new ( self ) : <EOL> for f in self . TYPES [ self . type ] : <EOL> v = self . _ask_field_value ( f ) <EOL> keyring . set_password ( self . uid , f , v ) <EOL> pass <EOL> @ property <EOL> def uid ( self ) : <EOL> return "<STR_LIT>" % ( self . name ) <EOL> @ property <EOL> def is_known ( self ) : <EOL> uid = self . uid <EOL> try : <EOL> return all ( keyring . get_password ( uid , f ) is not None for f in self . TYPES [ self . type ] ) <EOL> except Exception as exc : <EOL> lgr . warning ( "<STR_LIT>" % exc_str ( exc ) ) <EOL> return False <EOL> def __call__ ( self ) : <EOL> uid = self . uid <EOL> credentials = { } <EOL> for f in self . TYPES [ self . type ] : <EOL> v = keyring . get_password ( uid , f ) <EOL> while v is None : <EOL> v = self . _ask_field_value ( f ) <EOL> keyring . set_password ( uid , f , v ) <EOL> credentials [ f ] = v <EOL> return credentials <EOL> from . http import HTTPDownloader <EOL> @ auto_repr <EOL> class Provider ( object ) : <EOL> """<STR_LIT>""" <EOL> DOWNLOADERS = { '<STR_LIT:http>' : HTTPDownloader , <EOL> '<STR_LIT>' : HTTPDownloader , <EOL> '<STR_LIT>' : S3Downloader , <EOL> } <EOL> def __init__ ( self , name , url_res , credential = None , authenticator = None , <EOL> downloader = None ) : <EOL> """<STR_LIT>""" <EOL> self . name = name <EOL> self . url_res = assure_list_from_str ( url_res ) <EOL> self . credential = credential <EOL> self . authenticator = authenticator <EOL> self . _downloader = downloader <EOL> @ property <EOL> def downloader ( self ) : <EOL> return self . _downloader <EOL> @ staticmethod <EOL> def get_scheme_from_url ( url ) : <EOL> """<STR_LIT>""" <EOL> url_split = urlparse ( url ) <EOL> return url_split . scheme <EOL> @ classmethod <EOL> def _get_downloader_class ( cls , url ) : <EOL> key = cls . get_scheme_from_url ( url ) <EOL> if key in cls . DOWNLOADERS : <EOL> return cls . DOWNLOADERS [ key ] <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % ( url , key , cls . DOWNLOADERS . keys ( ) ) ) <EOL> def get_downloader ( self , url ) : <EOL> """<STR_LIT>""" <EOL> if self . _downloader is None : <EOL> Downloader = self . _get_downloader_class ( url ) <EOL> kwargs = { } <EOL> if self . credential : <EOL> kwargs [ '<STR_LIT>' ] = self . credential <EOL> if self . authenticator : <EOL> kwargs [ '<STR_LIT>' ] = self . authenticator <EOL> self . _downloader = Downloader ( ** kwargs ) <EOL> return self . _downloader <EOL> class Providers ( object ) : <EOL> """<STR_LIT>""" <EOL> AUTHENTICATION_TYPES = { <EOL> '<STR_LIT>' : HTMLFormAuthenticator , <EOL> '<STR_LIT>' : HTTPBasicAuthAuthenticator , <EOL> '<STR_LIT>' : HTTPBasicAuthAuthenticator , <EOL> '<STR_LIT>' : HTTPDigestAuthAuthenticator , <EOL> '<STR_LIT>' : S3Authenticator , <EOL> '<STR_LIT>' : NotImplementedAuthenticator , <EOL> '<STR_LIT:none>' : NoneAuthenticator , <EOL> } <EOL> CREDENTIAL_TYPES = { <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> } <EOL> _DEFAULT_PROVIDERS = None <EOL> def __init__ ( self , providers = None ) : <EOL> """<STR_LIT:U+0020>""" <EOL> self . _providers = providers or [ ] <EOL> self . _default_providers = { } <EOL> def __repr__ ( self ) : <EOL> return "<STR_LIT>" % ( self . __class__ . __name__ , "<STR_LIT>" if not self . _providers else repr ( self . _providers ) ) <EOL> def __len__ ( self ) : <EOL> return len ( self . _providers ) <EOL> def __getitem__ ( self , index ) : <EOL> return self . _providers [ index ] <EOL> def __iter__ ( self ) : <EOL> return self . _providers . __iter__ ( ) <EOL> @ classmethod <EOL> def from_config_files ( cls , files = None , reload = False ) : <EOL> """<STR_LIT>""" <EOL> if files is None and cls . _DEFAULT_PROVIDERS and not reload : <EOL> return cls . _DEFAULT_PROVIDERS <EOL> config = SafeConfigParserWithIncludes ( ) <EOL> files_orig = files <EOL> if files is None : <EOL> files = glob ( pathjoin ( dirname ( abspath ( __file__ ) ) , '<STR_LIT>' , '<STR_LIT>' ) ) <EOL> config . read ( files ) <EOL> providers = { } <EOL> credentials = { } <EOL> for section in config . sections ( ) : <EOL> if '<STR_LIT::>' in section : <EOL> type_ , name = section . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> assert ( type_ in { '<STR_LIT>' , '<STR_LIT>' } ) <EOL> items = { <EOL> o : config . get ( section , o ) for o in config . options ( section ) <EOL> } <EOL> locals ( ) . get ( type_ + "<STR_LIT:s>" ) [ name ] = getattr ( cls , '<STR_LIT>' + type_ ) ( name , items ) <EOL> if len ( items ) : <EOL> raise ValueError ( "<STR_LIT>" % ( name , str ( items ) ) ) <EOL> else : <EOL> lgr . warning ( "<STR_LIT>" % section ) <EOL> lgr . debug ( "<STR_LIT>" % len ( providers ) ) <EOL> for provider in providers . values ( ) : <EOL> if provider . credential : <EOL> if provider . credential not in credentials : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % ( provider . credential , "<STR_LIT:U+002CU+0020>" . join ( credentials . keys ( ) ) ) ) <EOL> provider . credential = credentials [ provider . credential ] <EOL> providers = Providers ( list ( providers . values ( ) ) ) <EOL> if files_orig is None : <EOL> cls . _DEFAULT_PROVIDERS = providers <EOL> return providers <EOL> @ classmethod <EOL> def reset_default_providers ( cls ) : <EOL> """<STR_LIT>""" <EOL> cls . _DEFAULT_PROVIDERS = None <EOL> @ classmethod <EOL> def _process_provider ( cls , name , items ) : <EOL> """<STR_LIT>""" <EOL> assert '<STR_LIT>' in items , "<STR_LIT>" <EOL> auth_type = items . pop ( '<STR_LIT>' ) <EOL> if auth_type not in cls . AUTHENTICATION_TYPES : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % ( auth_type , '<STR_LIT:U+002CU+0020>' . join ( cls . AUTHENTICATION_TYPES ) ) ) <EOL> if auth_type != '<STR_LIT:none>' : <EOL> authenticator = cls . AUTHENTICATION_TYPES [ auth_type ] ( <EOL> ** { k [ len ( auth_type ) + <NUM_LIT:1> : ] : items . pop ( k ) <EOL> for k in list ( items . keys ( ) ) <EOL> if k . startswith ( auth_type + "<STR_LIT:_>" ) } <EOL> ) <EOL> else : <EOL> authenticator = None <EOL> url_res = assure_list_from_str ( items . pop ( '<STR_LIT>' ) ) <EOL> assert url_res , "<STR_LIT>" <EOL> credential = items . pop ( '<STR_LIT>' , None ) <EOL> return Provider ( name = name , url_res = url_res , authenticator = authenticator , <EOL> credential = credential ) <EOL> @ classmethod <EOL> def _process_credential ( cls , name , items ) : <EOL> assert '<STR_LIT:type>' in items , "<STR_LIT>" % name <EOL> cred_type = items . pop ( '<STR_LIT:type>' ) <EOL> return Credential ( name = name , type = cred_type , url = items . pop ( '<STR_LIT:url>' , None ) ) <EOL> def get_provider ( self , url , only_nondefault = False ) : <EOL> """<STR_LIT>""" <EOL> nproviders = len ( self . _providers ) <EOL> for i in range ( nproviders ) : <EOL> provider = self . _providers [ i ] <EOL> if not provider . url_res : <EOL> continue <EOL> for url_re in provider . url_res : <EOL> if re . match ( url_re , url ) : <EOL> if i != <NUM_LIT:0> : <EOL> del self . _providers [ i ] <EOL> self . _providers = [ provider ] + self . _providers <EOL> assert ( len ( self . _providers ) == nproviders ) <EOL> lgr . debug ( "<STR_LIT>" , provider , url ) <EOL> return provider <EOL> if only_nondefault : <EOL> return None <EOL> scheme = Provider . get_scheme_from_url ( url ) <EOL> if scheme not in self . _default_providers : <EOL> lgr . debug ( "<STR_LIT>" % scheme ) <EOL> self . _default_providers [ scheme ] = Provider ( name = "<STR_LIT>" , url_res = [ "<STR_LIT>" % scheme ] ) <EOL> lgr . debug ( "<STR_LIT>" % scheme ) <EOL> return self . _default_providers [ scheme ] <EOL> def download ( self , url , * args , ** kwargs ) : <EOL> return self . get_provider ( url ) . get_downloader ( url ) . download ( url , * args , ** kwargs ) <EOL> def fetch ( self , url , * args , ** kwargs ) : <EOL> return self . get_provider ( url ) . get_downloader ( url ) . fetch ( url , * args , ** kwargs ) <EOL> def get_status ( self , url , * args , ** kwargs ) : <EOL> return self . get_provider ( url ) . get_downloader ( url ) . get_status ( url , * args , ** kwargs ) <EOL> def needs_authentication ( self , url ) : <EOL> provider = self . get_provider ( url , only_nondefault = True ) <EOL> if provider is None : <EOL> return None <EOL> return provider . authenticator is not None </s>
