def _get_client_builder()

in client/python/cli/polaris_cli.py [0:0]

• 47 lines of code
• 25 McCabe index (conditional complexity)

    def _get_client_builder(options):
        profile = {}
        client_profile = options.profile or os.getenv(CLIENT_PROFILE_ENV)
        if client_profile:
            profiles = PolarisCli._load_profiles()
            profile = profiles.get(client_profile)
            if not profile:
                raise Exception(f'Polaris profile {client_profile} not found')
        # Determine which credentials to use
        client_id = options.client_id or os.getenv(CLIENT_ID_ENV) or profile.get('client_id')
        client_secret = options.client_secret or os.getenv(CLIENT_SECRET_ENV) or profile.get('client_secret')
        
        # Validates
        has_access_token = options.access_token is not None
        has_client_secret = client_id is not None and client_secret is not None
        if has_access_token and (options.client_id or options.client_secret):
            raise Exception(f'Please provide credentials via either {Argument.to_flag_name(Arguments.CLIENT_ID)} &'
                            f' {Argument.to_flag_name(Arguments.CLIENT_SECRET)} or'
                            f' {Argument.to_flag_name(Arguments.ACCESS_TOKEN)}, but not both')
        if not has_access_token and not has_client_secret:
            raise Exception(f'Please provide credentials via either {Argument.to_flag_name(Arguments.CLIENT_ID)} &'
                            f' {Argument.to_flag_name(Arguments.CLIENT_SECRET)} or'
                            f' {Argument.to_flag_name(Arguments.ACCESS_TOKEN)}.'
                            f' Alternatively, you may set the environment variables {CLIENT_ID_ENV} &'
                            f' {CLIENT_SECRET_ENV}.')
        # Authenticate accordingly
        if options.base_url:
            if options.host is not None or options.port is not None:
                raise Exception(f'Please provide either {Argument.to_flag_name(Arguments.BASE_URL)} or'
                                f' {Argument.to_flag_name(Arguments.HOST)} &'
                                f' {Argument.to_flag_name(Arguments.PORT)}, but not both');

            polaris_management_url = f'{options.base_url}/api/management/v1'
            polaris_catalog_url = f'{options.base_url}/api/catalog/v1'
        else:
            host = options.host or profile.get('host') or DEFAULT_HOSTNAME
            port = options.port or profile.get('port') or DEFAULT_PORT
            polaris_management_url = f'http://{host}:{port}/api/management/v1'
            polaris_catalog_url = f'http://{host}:{port}/api/catalog/v1'

        config = Configuration(host=polaris_management_url)
        config.proxy = options.proxy
        if has_access_token:
            config.access_token = options.access_token
        elif has_client_secret:
            config.username = client_id
            config.password = client_secret

        if not has_access_token and not PolarisCli.DIRECT_AUTHENTICATION_ENABLED:
            token = PolarisCli._get_token(ApiClient(config), polaris_catalog_url, client_id, client_secret)
            config.username = None
            config.password = None
            config.access_token = token

        return lambda: ApiClient(config)