<s> """<STR_LIT>""" <EOL> import logging <EOL> try : <EOL> import urlparse <EOL> except ImportError : <EOL> import urllib . parse as urlparse <EOL> try : <EOL> import ssl <EOL> except ImportError : <EOL> ssl = None <EOL> from amqpstorm import compatibility <EOL> from amqpstorm . connection import Connection <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> class UriConnection ( Connection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , uri , lazy = False ) : <EOL> """<STR_LIT>""" <EOL> uri = compatibility . patch_uri ( uri ) <EOL> parsed_uri = urlparse . urlparse ( uri ) <EOL> use_ssl = parsed_uri . scheme == '<STR_LIT>' <EOL> hostname = parsed_uri . hostname or '<STR_LIT:localhost>' <EOL> port = parsed_uri . port or <NUM_LIT> <EOL> username = parsed_uri . username or '<STR_LIT>' <EOL> password = parsed_uri . password or '<STR_LIT>' <EOL> kwargs = self . _parse_uri_options ( parsed_uri , use_ssl , lazy ) <EOL> super ( UriConnection , self ) . __init__ ( hostname , username , <EOL> password , port , <EOL> ** kwargs ) <EOL> def _parse_uri_options ( self , parsed_uri , use_ssl , lazy ) : <EOL> """<STR_LIT>""" <EOL> kwargs = urlparse . parse_qs ( parsed_uri . query ) <EOL> options = { <EOL> '<STR_LIT>' : use_ssl , <EOL> '<STR_LIT>' : urlparse . unquote ( parsed_uri . path [ <NUM_LIT:1> : ] ) or '<STR_LIT:/>' , <EOL> '<STR_LIT>' : int ( kwargs . get ( '<STR_LIT>' , [ <NUM_LIT> ] ) [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' : int ( kwargs . get ( '<STR_LIT>' , [ <NUM_LIT:30> ] ) [ <NUM_LIT:0> ] ) , <EOL> '<STR_LIT>' : lazy <EOL> } <EOL> if compatibility . SSL_SUPPORTED and use_ssl : <EOL> options [ '<STR_LIT>' ] = self . _parse_ssl_options ( kwargs ) <EOL> return options <EOL> def _parse_ssl_options ( self , ssl_kwargs ) : <EOL> """<STR_LIT>""" <EOL> ssl_options = { } <EOL> for key in ssl_kwargs : <EOL> if key not in compatibility . SSL_OPTIONS : <EOL> continue <EOL> if '<STR_LIT>' in key : <EOL> value = self . _get_ssl_version ( ssl_kwargs [ key ] [ <NUM_LIT:0> ] ) <EOL> elif '<STR_LIT>' in key : <EOL> value = self . _get_ssl_validation ( ssl_kwargs [ key ] [ <NUM_LIT:0> ] ) <EOL> else : <EOL> value = ssl_kwargs [ key ] [ <NUM_LIT:0> ] <EOL> ssl_options [ key ] = value <EOL> return ssl_options <EOL> def _get_ssl_version ( self , value ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_ssl_attribute ( value , compatibility . SSL_VERSIONS , <EOL> ssl . PROTOCOL_TLSv1 , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def _get_ssl_validation ( self , value ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_ssl_attribute ( value , compatibility . SSL_CERT_MAP , <EOL> ssl . CERT_NONE , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> @ staticmethod <EOL> def _get_ssl_attribute ( value , mapping , default_value , warning_message ) : <EOL> """<STR_LIT>""" <EOL> for key in mapping : <EOL> if not key . endswith ( value . lower ( ) ) : <EOL> continue <EOL> return mapping [ key ] <EOL> LOGGER . warning ( warning_message , value ) <EOL> return default_value </s>
