def __init__()

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

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

    def __init__(self, config: NodeConfig):
        if not _REQUESTS_AVAILABLE:  # pragma: nocover
            raise ValueError(
                "You must have 'requests' installed to use RequestsHttpNode"
            )

        super().__init__(config)

        # Initialize Session so .headers works before calling super().__init__().
        self.session = requests.Session()
        self.session.headers.clear()  # Empty out all the default session headers

        if config.scheme == "https":
            # If we're using ssl_assert_fingerprint we don't want
            # to verify certificates the typical way. Instead we
            # rely on the custom ElasticHTTPAdapter and urllib3.
            if config.ssl_assert_fingerprint:
                self.session.verify = False

            # Otherwise we go the traditional route of verifying certs.
            else:
                if config.ca_certs:
                    if not config.verify_certs:
                        raise ValueError(
                            "You cannot use 'ca_certs' when 'verify_certs=False'"
                        )
                    self.session.verify = config.ca_certs
                else:
                    self.session.verify = config.verify_certs

                if not config.ssl_show_warn:
                    urllib3.disable_warnings()

                if (
                    config.scheme == "https"
                    and not config.verify_certs
                    and 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,
                    )

        # Requests supports setting 'session.auth' via _extras['requests.session.auth'] = ...
        try:
            requests_session_auth: Optional[AuthBase] = config._extras.pop(
                "requests.session.auth", None
            )
        except AttributeError:
            requests_session_auth = None
        if requests_session_auth is not None:
            self.session.auth = requests_session_auth

        # Client certificates
        if config.client_cert:
            if config.client_key:
                self.session.cert = (config.client_cert, config.client_key)
            else:
                self.session.cert = config.client_cert

        # Create and mount custom adapter for constraining number of connections
        adapter = _ElasticHTTPAdapter(
            node_config=config,
            pool_connections=config.connections_per_node,
            pool_maxsize=config.connections_per_node,
            pool_block=True,
        )
        # Preload the HTTPConnectionPool so initialization issues
        # are raised here instead of in perform_request()
        if hasattr(adapter, "get_connection_with_tls_context"):
            request = requests.Request(method="GET", url=self.base_url)
            prepared_request = self.session.prepare_request(request)
            adapter.get_connection_with_tls_context(
                prepared_request, verify=self.session.verify
            )
        else:
            # elastic-transport is not vulnerable to CVE-2024-35195 because it uses
            # requests.Session and an SSLContext without using the verify parameter.
            # We should remove this branch when requiring requests 2.32 or later.
            adapter.get_connection(self.base_url)

        self.session.mount(prefix=f"{self.scheme}://", adapter=adapter)