def __init__()

in azure-kusto-data/azure/kusto/data/security.py [0:0]

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

    def __init__(self, kcsb: "KustoConnectionStringBuilder", is_async: bool):
        parsed_url = urlparse(kcsb.data_source)
        self.kusto_uri = f"{parsed_url.scheme}://{parsed_url.hostname}"
        if parsed_url.port is not None:
            self.kusto_uri += f":{parsed_url.port}"

        self.username = None

        if kcsb.interactive_login:
            self.token_provider = InteractiveLoginTokenProvider(self.kusto_uri, kcsb.authority_id, kcsb.login_hint, kcsb.domain_hint, is_async=is_async)
        elif all([kcsb.aad_user_id, kcsb.password]):
            self.token_provider = UserPassTokenProvider(self.kusto_uri, kcsb.authority_id, kcsb.aad_user_id, kcsb.password, is_async=is_async)
        elif all([kcsb.application_client_id, kcsb.application_key]):
            self.token_provider = ApplicationKeyTokenProvider(
                self.kusto_uri, kcsb.authority_id, kcsb.application_client_id, kcsb.application_key, is_async=is_async
            )
        elif all([kcsb.application_client_id, kcsb.application_certificate, kcsb.application_certificate_thumbprint]):
            # kcsb.application_public_certificate can be None if SNI is not used
            self.token_provider = ApplicationCertificateTokenProvider(
                self.kusto_uri,
                kcsb.application_client_id,
                kcsb.authority_id,
                kcsb.application_certificate,
                kcsb.application_certificate_thumbprint,
                kcsb.application_public_certificate,
                is_async=is_async,
            )
        elif kcsb.msi_authentication:
            self.token_provider = MsiTokenProvider(self.kusto_uri, kcsb.msi_parameters, is_async=is_async)
        elif kcsb.user_token:
            self.token_provider = BasicTokenProvider(kcsb.user_token, is_async=is_async)
        elif kcsb.application_token:
            self.token_provider = BasicTokenProvider(kcsb.application_token, is_async=is_async)
        elif kcsb.az_cli_login:
            self.token_provider = AzCliTokenProvider(self.kusto_uri, is_async=is_async)
        elif kcsb.token_provider or kcsb.async_token_provider:
            self.token_provider = CallbackTokenProvider(token_callback=kcsb.token_provider, async_token_callback=kcsb.async_token_provider, is_async=is_async)
        elif kcsb.token_credential_login:
            self.token_provider = AzureIdentityTokenCredentialProvider(
                self.kusto_uri,
                is_async=is_async,
                credential=kcsb.azure_credential,
                credential_from_login_endpoint=kcsb.azure_credential_from_login_endpoint,
            )
        elif kcsb.device_login:
            self.token_provider = DeviceLoginTokenProvider(self.kusto_uri, kcsb.authority_id, kcsb.device_callback, is_async=is_async)
        else:
            self.token_provider = InteractiveLoginTokenProvider(self.kusto_uri, kcsb.authority_id, kcsb.login_hint, kcsb.domain_hint, is_async=is_async)