def __init__()

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

• 20 lines of code
• 6 McCabe index (conditional complexity)

    def __init__(self, kcsb: Union[KustoConnectionStringBuilder, str], is_async):
        self._kcsb = kcsb
        self._proxy_url: Optional[str] = None
        if not isinstance(kcsb, KustoConnectionStringBuilder):
            self._kcsb = KustoConnectionStringBuilder(kcsb)
        self._kusto_cluster = self._kcsb.data_source

        # notice that in this context, federated actually just stands for aad auth, not aad federated auth (legacy code)
        self._aad_helper = _AadHelper(self._kcsb, is_async) if self._kcsb.aad_federated_security else None

        if not self._kusto_cluster.endswith("/"):
            self._kusto_cluster += "/"

        # Create a session object for connection pooling
        self._mgmt_endpoint = urljoin(self._kusto_cluster, "v1/rest/mgmt")
        self._query_endpoint = urljoin(self._kusto_cluster, "v2/rest/query")
        self._streaming_ingest_endpoint = urljoin(self._kusto_cluster, "v1/rest/ingest/")
        self._request_headers = {
            "Accept": "application/json",
            "Accept-Encoding": "gzip,deflate",
            "x-ms-version": self.API_VERSION,
        }

        self.client_details = self._kcsb.client_details
        self._is_closed: bool = False

        self.default_database = self._kcsb.initial_catalog