def __init__()

in core/kibana.py [0:0]

• 31 lines of code
• 11 McCabe index (conditional complexity)

    def __init__(self, url=None, cloud_id=None, basic_auth=None, api_key=None, verify_certs=True, ca_certs=None):
        if not (url or cloud_id):
            raise ValueError("Either `url` or `cloud_id` must be defined")

        self.url = url
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json", "kbn-xsrf": str(uuid.uuid4())})

        if api_key:
            self.session.headers["Authorization"] = f"ApiKey {api_key}"
        if basic_auth:
            self.session.auth = requests.auth.HTTPBasicAuth(*basic_auth)
        if not verify_certs:
            self.session.verify = False
        elif ca_certs:
            self.session.verify = ca_certs

        retry_strategy = requests.packages.urllib3.util.retry.Retry(
            total=3,
            allowed_methods=["HEAD", "GET", "OPTIONS"],
        )

        adapter = requests.adapters.HTTPAdapter(max_retries=retry_strategy)
        self.session.mount("https://", adapter)
        self.session.mount("http://", adapter)

        if cloud_id:
            import base64

            cluster_name, cloud_info = cloud_id.split(":")
            domain, es_uuid, kibana_uuid = base64.b64decode(cloud_info.encode("utf-8")).decode("utf-8").split("$")

            if domain.endswith(":443"):
                domain = domain[:-4]

            url_from_cloud = f"https://{kibana_uuid}.{domain}:9243"
            if self.url and self.url != url_from_cloud:
                raise ValueError(f"url provided ({self.url}) does not match url derived from cloud_id {url_from_cloud}")
            self.url = url_from_cloud