<s> """<STR_LIT>""" <EOL> import warnings <EOL> from bson . py3compat import PY3 , iteritems , string_type <EOL> if PY3 : <EOL> from urllib . parse import unquote_plus <EOL> else : <EOL> from urllib import unquote_plus <EOL> from pymongo . common import get_validated_options <EOL> from pymongo . errors import ConfigurationError , InvalidURI <EOL> SCHEME = '<STR_LIT>' <EOL> SCHEME_LEN = len ( SCHEME ) <EOL> DEFAULT_PORT = <NUM_LIT> <EOL> def _partition ( entity , sep ) : <EOL> """<STR_LIT>""" <EOL> parts = entity . split ( sep , <NUM_LIT:1> ) <EOL> if len ( parts ) == <NUM_LIT:2> : <EOL> return parts [ <NUM_LIT:0> ] , sep , parts [ <NUM_LIT:1> ] <EOL> else : <EOL> return entity , '<STR_LIT>' , '<STR_LIT>' <EOL> def _rpartition ( entity , sep ) : <EOL> """<STR_LIT>""" <EOL> idx = entity . rfind ( sep ) <EOL> if idx == - <NUM_LIT:1> : <EOL> return '<STR_LIT>' , '<STR_LIT>' , entity <EOL> return entity [ : idx ] , sep , entity [ idx + <NUM_LIT:1> : ] <EOL> def parse_userinfo ( userinfo ) : <EOL> """<STR_LIT>""" <EOL> if '<STR_LIT:@>' in userinfo or userinfo . count ( '<STR_LIT::>' ) > <NUM_LIT:1> : <EOL> raise InvalidURI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> user , _ , passwd = _partition ( userinfo , "<STR_LIT::>" ) <EOL> if not user : <EOL> raise InvalidURI ( "<STR_LIT>" ) <EOL> user = unquote_plus ( user ) <EOL> passwd = unquote_plus ( passwd ) <EOL> return user , passwd <EOL> def parse_ipv6_literal_host ( entity , default_port ) : <EOL> """<STR_LIT>""" <EOL> if entity . find ( '<STR_LIT:]>' ) == - <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> i = entity . find ( '<STR_LIT>' ) <EOL> if i == - <NUM_LIT:1> : <EOL> return entity [ <NUM_LIT:1> : - <NUM_LIT:1> ] , default_port <EOL> return entity [ <NUM_LIT:1> : i ] , entity [ i + <NUM_LIT:2> : ] <EOL> def parse_host ( entity , default_port = DEFAULT_PORT ) : <EOL> """<STR_LIT>""" <EOL> host = entity <EOL> port = default_port <EOL> if entity [ <NUM_LIT:0> ] == '<STR_LIT:[>' : <EOL> host , port = parse_ipv6_literal_host ( entity , default_port ) <EOL> elif entity . endswith ( "<STR_LIT>" ) : <EOL> return entity , default_port <EOL> elif entity . find ( '<STR_LIT::>' ) != - <NUM_LIT:1> : <EOL> if entity . count ( '<STR_LIT::>' ) > <NUM_LIT:1> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> host , port = host . split ( '<STR_LIT::>' , <NUM_LIT:1> ) <EOL> if isinstance ( port , string_type ) : <EOL> if not port . isdigit ( ) or int ( port ) > <NUM_LIT> or int ( port ) <= <NUM_LIT:0> : <EOL> raise ValueError ( "<STR_LIT>" <EOL> % ( port , ) ) <EOL> port = int ( port ) <EOL> return host . lower ( ) , port <EOL> def validate_options ( opts , warn = False ) : <EOL> """<STR_LIT>""" <EOL> return get_validated_options ( opts , warn ) <EOL> def _parse_options ( opts , delim ) : <EOL> """<STR_LIT>""" <EOL> options = { } <EOL> for opt in opts . split ( delim ) : <EOL> key , val = opt . split ( "<STR_LIT:=>" ) <EOL> if key . lower ( ) == '<STR_LIT>' : <EOL> options . setdefault ( '<STR_LIT>' , [ ] ) . append ( val ) <EOL> else : <EOL> if str ( key ) in options : <EOL> warnings . warn ( "<STR_LIT>" % ( str ( key ) , ) ) <EOL> options [ str ( key ) ] = unquote_plus ( val ) <EOL> if "<STR_LIT>" in options : <EOL> if "<STR_LIT>" in options : <EOL> options . pop ( "<STR_LIT>" ) <EOL> warnings . warn ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return options <EOL> def split_options ( opts , validate = True , warn = False ) : <EOL> """<STR_LIT>""" <EOL> and_idx = opts . find ( "<STR_LIT:&>" ) <EOL> semi_idx = opts . find ( "<STR_LIT:;>" ) <EOL> try : <EOL> if and_idx >= <NUM_LIT:0> and semi_idx >= <NUM_LIT:0> : <EOL> raise InvalidURI ( "<STR_LIT>" ) <EOL> elif and_idx >= <NUM_LIT:0> : <EOL> options = _parse_options ( opts , "<STR_LIT:&>" ) <EOL> elif semi_idx >= <NUM_LIT:0> : <EOL> options = _parse_options ( opts , "<STR_LIT:;>" ) <EOL> elif opts . find ( "<STR_LIT:=>" ) != - <NUM_LIT:1> : <EOL> options = _parse_options ( opts , None ) <EOL> else : <EOL> raise ValueError <EOL> except ValueError : <EOL> raise InvalidURI ( "<STR_LIT>" ) <EOL> if validate : <EOL> return validate_options ( options , warn ) <EOL> return options <EOL> def split_hosts ( hosts , default_port = DEFAULT_PORT ) : <EOL> """<STR_LIT>""" <EOL> nodes = [ ] <EOL> for entity in hosts . split ( '<STR_LIT:U+002C>' ) : <EOL> if not entity : <EOL> raise ConfigurationError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> port = default_port <EOL> if entity . endswith ( '<STR_LIT>' ) : <EOL> port = None <EOL> nodes . append ( parse_host ( entity , port ) ) <EOL> return nodes <EOL> def parse_uri ( uri , default_port = DEFAULT_PORT , validate = True , warn = False ) : <EOL> """<STR_LIT>""" <EOL> if not uri . startswith ( SCHEME ) : <EOL> raise InvalidURI ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( SCHEME , ) ) <EOL> scheme_free = uri [ SCHEME_LEN : ] <EOL> if not scheme_free : <EOL> raise InvalidURI ( "<STR_LIT>" ) <EOL> user = None <EOL> passwd = None <EOL> dbase = None <EOL> collection = None <EOL> options = { } <EOL> if '<STR_LIT>' in scheme_free : <EOL> host_part , _ , path_part = _rpartition ( scheme_free , '<STR_LIT:/>' ) <EOL> if not host_part : <EOL> host_part = path_part <EOL> path_part = "<STR_LIT>" <EOL> if '<STR_LIT:/>' in host_part : <EOL> raise InvalidURI ( "<STR_LIT>" <EOL> "<STR_LIT>" % host_part ) <EOL> host_part = unquote_plus ( host_part ) <EOL> path_part = unquote_plus ( path_part ) <EOL> else : <EOL> host_part , _ , path_part = _partition ( scheme_free , '<STR_LIT:/>' ) <EOL> if not path_part and '<STR_LIT:?>' in host_part : <EOL> raise InvalidURI ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> if '<STR_LIT:@>' in host_part : <EOL> userinfo , _ , hosts = _rpartition ( host_part , '<STR_LIT:@>' ) <EOL> user , passwd = parse_userinfo ( userinfo ) <EOL> else : <EOL> hosts = host_part <EOL> nodes = split_hosts ( hosts , default_port = default_port ) <EOL> if path_part : <EOL> if path_part [ <NUM_LIT:0> ] == '<STR_LIT:?>' : <EOL> opts = path_part [ <NUM_LIT:1> : ] <EOL> else : <EOL> dbase , _ , opts = _partition ( path_part , '<STR_LIT:?>' ) <EOL> if '<STR_LIT:.>' in dbase : <EOL> dbase , collection = dbase . split ( '<STR_LIT:.>' , <NUM_LIT:1> ) <EOL> if opts : <EOL> options = split_options ( opts , validate , warn ) <EOL> if dbase is not None : <EOL> dbase = unquote_plus ( dbase ) <EOL> if collection is not None : <EOL> collection = unquote_plus ( collection ) <EOL> return { <EOL> '<STR_LIT>' : nodes , <EOL> '<STR_LIT:username>' : user , <EOL> '<STR_LIT:password>' : passwd , <EOL> '<STR_LIT>' : dbase , <EOL> '<STR_LIT>' : collection , <EOL> '<STR_LIT>' : options <EOL> } <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import pprint <EOL> import sys <EOL> try : <EOL> pprint . pprint ( parse_uri ( sys . argv [ <NUM_LIT:1> ] ) ) <EOL> except InvalidURI as e : <EOL> print ( e ) <EOL> sys . exit ( <NUM_LIT:0> ) </s>
