def _process_args()

in python-phoenixdb/phoenixdb/__init__.py [0:0]

• 48 lines of code
• 37 McCabe index (conditional complexity)

def _process_args(
        url, auth=None, authentication=None, avatica_user=None, avatica_password=None,
        truststore=None, verify=None, do_as=None, user=None, password=None):
    url_parsed = urlparse(url)
    url_params = parse_qs(url_parsed.query, keep_blank_values=True)

    # Parse supported JDBC compatible parameters from URL. args have precendece
    # Unlike the JDBC driver, we are expecting these as query params, as the avatica java client
    # has a different idea of what an URL param is than urlparse. (urlparse seems just broken
    # in this regard)
    params_changed = False
    if auth is None and authentication is None and 'authentication' in url_params:
        authentication = url_params['authentication'][0]
        del url_params['authentication']
        params_changed = True

    if avatica_user is None and 'avatica_user' in url_params:
        avatica_user = url_params['avatica_user'][0]
        del url_params['avatica_user']
        params_changed = True

    if avatica_password is None and 'avatica_password' in url_params:
        avatica_password = url_params['avatica_password'][0]
        del url_params['avatica_password']
        params_changed = True

    if verify is None and truststore is None and 'truststore' in url_params:
        truststore = url_params['truststore'][0]
        del url_params['truststore']
        params_changed = True

    if authentication == 'BASIC' or authentication == 'DIGEST':
        # Handle standard user and password parameters
        if user is not None and avatica_user is None:
            avatica_user = user
        if password is not None and avatica_password is None:
            avatica_password = password
    else:
        # interpret standard user parameter as do_as for SPNEGO and NONE
        if user is not None and do_as is None:
            do_as = user

    # Add doAs
    if do_as:
        url_params['doAs'] = do_as
        params_changed = True

    if params_changed:
        url_parsed = url_parsed._replace(query=urlencode(url_params))
        url = urlunparse(url_parsed)

    if auth == "SPNEGO":
        # Special case for backwards compatibility
        auth = _get_SPNEGOAuth()
    elif auth is None and authentication is not None:
        if authentication == "SPNEGO":
            auth = _get_SPNEGOAuth()
        elif authentication == "BASIC" and avatica_user is not None and avatica_password is not None:
            auth = HTTPBasicAuth(avatica_user, avatica_password)
        elif authentication == "DIGEST" and avatica_user is not None and avatica_password is not None:
            auth = HTTPDigestAuth(avatica_user, avatica_password)

    if verify is None and truststore is not None:
        verify = truststore

    return (url, auth, verify)