def __init__()

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

• 52 lines of code
• 22 McCabe index (conditional complexity)

    def __init__(self, config: NodeConfig):
        if not _HTTPX_AVAILABLE:  # pragma: nocover
            raise ValueError("You must have 'httpx' installed to use HttpxNode")
        super().__init__(config)

        if config.ssl_assert_fingerprint:
            raise ValueError(
                "httpx does not support certificate pinning. https://github.com/encode/httpx/issues/761"
            )

        ssl_context: Union[ssl.SSLContext, Literal[False]] = False
        if config.scheme == "https":
            if config.ssl_context is not None:
                ssl_context = ssl_context_from_node_config(config)
            else:
                ssl_context = ssl_context_from_node_config(config)

                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."
                        )
                else:
                    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,
                        )

                if ca_certs is not None:
                    if os.path.isfile(ca_certs):
                        ssl_context.load_verify_locations(cafile=ca_certs)
                    elif os.path.isdir(ca_certs):
                        ssl_context.load_verify_locations(capath=ca_certs)
                    else:
                        raise ValueError("ca_certs parameter is not a path")

                # Use client_cert and client_key variables for SSL certificate configuration.
                if config.client_cert and not os.path.isfile(config.client_cert):
                    raise ValueError("client_cert is not a path to a file")
                if config.client_key and not os.path.isfile(config.client_key):
                    raise ValueError("client_key is not a path to a file")
                if config.client_cert and config.client_key:
                    ssl_context.load_cert_chain(config.client_cert, config.client_key)
                elif config.client_cert:
                    ssl_context.load_cert_chain(config.client_cert)

        self.client = httpx.AsyncClient(
            base_url=f"{config.scheme}://{config.host}:{config.port}",
            limits=httpx.Limits(max_connections=config.connections_per_node),
            verify=ssl_context or False,
            timeout=config.request_timeout,
        )