def __init__()

in elastic_transport/_node/_http_urllib3.py [0:0]

• 60 lines of code
• 9 McCabe index (conditional complexity)

    def __init__(self, config: NodeConfig):
        super().__init__(config)

        pool_class = urllib3.HTTPConnectionPool
        kw: Dict[str, Any] = {}

        if config.scheme == "https":
            pool_class = HTTPSConnectionPool
            ssl_context = ssl_context_from_node_config(config)
            kw["ssl_context"] = ssl_context

            if config.ssl_assert_hostname and config.ssl_assert_fingerprint:
                raise ValueError(
                    "Can't specify both 'ssl_assert_hostname' and 'ssl_assert_fingerprint'"
                )

            # Fingerprint verification doesn't require CA certificates being loaded.
            # We also want to disable other verification methods as we only care
            # about the fingerprint of the certificates, not whether they form
            # a verified chain to a trust anchor.
            elif config.ssl_assert_fingerprint:
                # Manually disable these in the right order on the SSLContext
                # so urllib3 won't think we want conflicting things.
                ssl_context.check_hostname = False
                ssl_context.verify_mode = ssl.CERT_NONE

                kw.update(
                    {
                        "assert_fingerprint": config.ssl_assert_fingerprint,
                        "assert_hostname": False,
                        "cert_reqs": "CERT_NONE",
                    }
                )

            else:
                kw["assert_hostname"] = config.ssl_assert_hostname

                # Convert all sentinel values to their actual default
                # values if not using an SSLContext.
                ca_certs = (
                    DEFAULT_CA_CERTS if config.ca_certs is None else config.ca_certs
                )
                if config.verify_certs:
                    if not ca_certs:
                        raise ValueError(
                            "Root certificates are missing for certificate "
                            "validation. Either pass them in using the ca_certs parameter or "
                            "install certifi to use it automatically."
                        )

                    kw.update(
                        {
                            "cert_reqs": "CERT_REQUIRED",
                            "ca_certs": ca_certs,
                            "cert_file": config.client_cert,
                            "key_file": config.client_key,
                        }
                    )
                else:
                    kw["cert_reqs"] = "CERT_NONE"

                    if config.ssl_show_warn:
                        warnings.warn(
                            f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure",
                            stacklevel=warn_stacklevel(),
                            category=SecurityWarning,
                        )
                    else:
                        urllib3.disable_warnings()

        self.pool = pool_class(
            config.host,
            port=config.port,
            timeout=urllib3.Timeout(total=config.request_timeout),
            maxsize=config.connections_per_node,
            block=True,
            **kw,
        )