def parse_pg_scheme_url()

in aws_advanced_python_wrapper/utils/properties.py [0:0]

• 56 lines of code
• 15 McCabe index (conditional complexity)

    def parse_pg_scheme_url(conn_info: str) -> Properties:
        props = Properties()
        if conn_info.startswith("postgresql://"):
            to_parse = conn_info[len("postgresql://"):]
        elif conn_info.startswith("postgres://"):
            to_parse = conn_info[len("postgres://"):]
        else:
            raise AwsWrapperError(Messages.get_formatted("PropertiesUtils.InvalidPgSchemeUrl", conn_info))

        # Example URL: postgresql://user:password@host:port/dbname?some_prop=some_value
        # More examples here: https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-CONNSTRING
        host_separator = to_parse.find("@")
        if host_separator >= 0:
            user_spec = to_parse[:host_separator]
            password_separator = user_spec.find(":")
            if password_separator >= 0:
                props[WrapperProperties.USER.name] = user_spec[:password_separator]
                props[WrapperProperties.PASSWORD.name] = user_spec[password_separator + 1:host_separator]
            else:
                props[WrapperProperties.USER.name] = user_spec
            to_parse = to_parse[host_separator + 1:]

        db_separator = to_parse.find("/")
        props_separator = to_parse.find("?")
        if db_separator >= 0:
            host_spec = to_parse[:db_separator]
            to_parse = to_parse[db_separator + 1:]
            props_separator = to_parse.find("?")
        elif props_separator >= 0:
            host_spec = to_parse[:props_separator]
            to_parse = to_parse[props_separator + 1:]
        else:
            host_spec = to_parse

        if host_spec.find(",") >= 0:
            raise AwsWrapperError(Messages.get_formatted("PropertiesUtils.MultipleHostsNotSupported", conn_info))

        # host_spec may be a percent-encoded unix domain socket, eg '%2Fvar%2Flib%2Fpostgresql'.
        # When stored as a kwarg instead of a connection string property, it should be decoded.
        host_spec = unquote(host_spec)
        if host_spec.startswith("["):
            # IPv6 addresses should be enclosed in square brackets, eg 'postgresql://[2001:db8::1234]/dbname'
            host_end = host_spec.find("]")
            props["host"] = host_spec[:host_end + 1]
            host_spec = host_spec[host_end + 1:]
            if len(host_spec) > 0:
                props["port"] = host_spec[1:]
        else:
            port_separator = host_spec.find(":")
            if port_separator >= 0:
                props["host"] = host_spec[:port_separator]
                props["port"] = host_spec[port_separator + 1:]
            else:
                if len(host_spec) > 0:
                    props["host"] = host_spec

        if db_separator >= 0:
            if props_separator >= 0:
                props[WrapperProperties.DATABASE.name] = to_parse[:props_separator]
                to_parse = to_parse[props_separator + 1:]
            else:
                props[WrapperProperties.DATABASE.name] = to_parse
                return props

        if props_separator >= 0:
            # Connection string properties must be percent-decoded when stored as kwargs
            props.update(PropertiesUtils.parse_key_values(to_parse, separator="&", percent_decode=True))
        return props