Sonar analysis

Author: kylstevenson

py_src/fusion/embeddings.py

@@ -285,6 +285,20 @@ def __init__(  # noqa: PLR0912, PLR0913, PLR0915
             **kwargs,
         )
 
+        self._configure_sync_connection(http_auth, verify_certs, ssl_show_warn, ca_certs, client_cert, client_key)
+        self.url_prefix, self.multi_dataset_url_prefix = _resolve_url_prefix(self.catalog, self.knowledge_base)
+
+        self.index_name: str | None = kwargs.get("index")
+
+    def _configure_sync_connection(
+        self,
+        http_auth: Any,
+        verify_certs: bool,
+        ssl_show_warn: bool,
+        ca_certs: Any,
+        client_cert: Any,
+        client_key: Any,
+    ) -> None:
         if not self.http_compress:
             # Need to set this to 'None' otherwise Requests adds its own.
             self.session.headers["accept-encoding"] = None  # type: ignore
@@ -313,9 +327,6 @@ def __init__(  # noqa: PLR0912, PLR0913, PLR0915
 
         if self.use_ssl and not verify_certs and ssl_show_warn:
             warnings.warn(f"Connecting to {self.host} using SSL with verify_certs=False is insecure.", stacklevel=2)
-        self.url_prefix, self.multi_dataset_url_prefix = _resolve_url_prefix(self.catalog, self.knowledge_base)
-
-        self.index_name: str | None = kwargs.get("index")
 
     def _tidy_url(self, url: str) -> str:
         return url.replace("%2F%7B", "/").replace("%7D%2F", "/").replace("%2F", "/")
@@ -576,40 +587,15 @@ def __init__(  # noqa: PLR0912, PLR0913, PLR0915
         )
 
         self.ssl_assert_fingerprint = ssl_assert_fingerprint
-        if self.use_ssl and ssl_context is None:
-            ssl_context = (
-                create_secure_ssl_context()
-                if ssl_version is None
-                else ssl.SSLContext(ssl_version)  # NOSONAR (S4423): preserve caller-supplied protocol
-            )
-
-            verify_certs, ssl_show_warn = _resolve_async_ssl_defaults(verify_certs, ssl_show_warn)
-
-            if verify_certs:
-                ssl_context.verify_mode = ssl.CERT_REQUIRED
-                ssl_context.check_hostname = True
-            else:
-                ssl_context.check_hostname = False
-                ssl_context.verify_mode = ssl.CERT_NONE
-
-            ca_certs = self.default_ca_certs() if ca_certs is None else ca_certs
-            if verify_certs:
-                if not ca_certs:
-                    raise ImproperlyConfigured(
-                        "Root certificates are missing for certificate "
-                        "validation. Either pass them in using the ca_certs parameter or "
-                        "install certifi to use it automatically."
-                    )
-                _load_verify_locations(ssl_context, ca_certs)
-            elif ssl_show_warn:
-                warnings.warn(f"Connecting to {self.host} using SSL with verify_certs=False is insecure.", stacklevel=2)
-
-            _validate_client_cert_path(client_cert, "client_cert")
-            _validate_client_cert_path(client_key, "client_key")
-            if client_cert and client_key:
-                ssl_context.load_cert_chain(client_cert, client_key)
-            elif client_cert:
-                ssl_context.load_cert_chain(client_cert)
+        ssl_context = self._build_async_ssl_context(
+            ssl_context,
+            verify_certs,
+            ssl_show_warn,
+            ca_certs,
+            client_cert,
+            client_key,
+            ssl_version,
+        )
 
         self.headers.setdefault("connection", "keep-alive")
         self.loop = loop
@@ -624,6 +610,159 @@ def __init__(  # noqa: PLR0912, PLR0913, PLR0915
         self._http_auth = http_auth
         self._ssl_context = ssl_context
 
+    def _build_async_ssl_context(
+        self,
+        ssl_context: Any,
+        verify_certs: Any,
+        ssl_show_warn: Any,
+        ca_certs: Any,
+        client_cert: Any,
+        client_key: Any,
+        ssl_version: Any,
+    ) -> Any:
+        if not self.use_ssl or ssl_context is not None:
+            return ssl_context
+
+        ssl_context = (
+            create_secure_ssl_context()
+            if ssl_version is None
+            else ssl.SSLContext(ssl_version)  # NOSONAR (S4423): preserve caller-supplied protocol
+        )
+        verify_certs, ssl_show_warn = _resolve_async_ssl_defaults(verify_certs, ssl_show_warn)
+        self._configure_async_ssl_context(ssl_context, verify_certs, ssl_show_warn, ca_certs)
+        self._load_async_client_cert_chain(ssl_context, client_cert, client_key)
+        return ssl_context
+
+    def _configure_async_ssl_context(
+        self,
+        ssl_context: ssl.SSLContext,
+        verify_certs: bool,
+        ssl_show_warn: bool,
+        ca_certs: Any,
+    ) -> None:
+        if verify_certs:
+            ssl_context.verify_mode = ssl.CERT_REQUIRED
+            ssl_context.check_hostname = True
+        else:
+            ssl_context.check_hostname = False
+            ssl_context.verify_mode = ssl.CERT_NONE
+
+        ca_certs = self.default_ca_certs() if ca_certs is None else ca_certs
+        if verify_certs:
+            if not ca_certs:
+                raise ImproperlyConfigured(
+                    "Root certificates are missing for certificate "
+                    "validation. Either pass them in using the ca_certs parameter or "
+                    "install certifi to use it automatically."
+                )
+            _load_verify_locations(ssl_context, ca_certs)
+        elif ssl_show_warn:
+            warnings.warn(f"Connecting to {self.host} using SSL with verify_certs=False is insecure.", stacklevel=2)
+
+    @staticmethod
+    def _load_async_client_cert_chain(
+        ssl_context: ssl.SSLContext,
+        client_cert: Any,
+        client_key: Any,
+    ) -> None:
+        _validate_client_cert_path(client_cert, "client_cert")
+        _validate_client_cert_path(client_key, "client_key")
+        if client_cert and client_key:
+            ssl_context.load_cert_chain(client_cert, client_key)
+        elif client_cert:
+            ssl_context.load_cert_chain(client_cert)
+
+    def _prepare_async_request(
+        self,
+        method: str,
+        url: str,
+        params: Mapping[str, Any] | None,
+        body: bytes | None,
+    ) -> tuple[str, str, bytes | None, bytes | None, str]:
+        method = _normalize_method(method)
+        url = self._tidy_url(url)
+        url = self._make_url_valid(url, body)
+        body = _prepare_embeddings_request_body(url, method, body, self.knowledge_base)
+        orig_body = body
+        query_string = urlencode(params) if params else ""
+        if query_string:
+            url = f"{url}?{query_string}"
+        return method, url, body, orig_body, query_string
+
+    def _build_async_request_headers(
+        self,
+        method: str,
+        url: str,
+        query_string: str,
+        body: bytes | None,
+        headers: Mapping[str, str] | None,
+    ) -> tuple[dict[str, Any], bytes | None, aiohttp.BasicAuth | None]:
+        req_headers = self.headers.copy()
+        if headers:
+            req_headers.update(headers)
+
+        if self.http_compress and body:
+            body = self._gzip_compress(body)
+            req_headers["content-encoding"] = "gzip"
+
+        auth = self._http_auth if isinstance(self._http_auth, aiohttp.BasicAuth) else None
+        if callable(self._http_auth):
+            req_headers = {
+                **req_headers,
+                **self._http_auth(method, url, query_string, body),
+            }
+        return req_headers, body, auth
+
+    def _raise_async_request_exception(
+        self,
+        method: str,
+        url: str,
+        orig_body: bytes | None,
+        start: float,
+        exception: Exception,
+    ) -> None:
+        self.log_request_fail(
+            method,
+            str(url),
+            url,
+            orig_body,
+            self.loop.time() - start,
+            exception=exception,
+        )
+        if isinstance(exception, aiohttp_exceptions.ServerFingerprintMismatch):
+            raise SSLError("N/A", str(exception), exception) from exception
+        if isinstance(exception, (asyncio.TimeoutError, aiohttp_exceptions.ServerTimeoutError)):
+            raise ConnectionTimeout("TIMEOUT", str(exception), exception) from exception
+        raise ConnectionError("N/A", str(exception), exception) from exception
+
+    def _handle_async_response(
+        self,
+        method: str,
+        url: str,
+        orig_body: bytes | None,
+        response: Any,
+        raw_data: str,
+        duration: float,
+        ignore: Collection[int],
+    ) -> None:
+        warning_headers = response.headers.getall("warning", ())
+        self._raise_warnings(warning_headers)
+
+        raw_data_modified = str(_modify_post_response_langchain(raw_data))
+        if not (HTTP_OK <= response.status < HTTP_MULTIPLE_CHOICES) and response.status not in ignore:
+            self.log_request_fail(
+                method,
+                str(url),
+                url,
+                orig_body,
+                duration,
+                status_code=response.status,
+                response=raw_data_modified,
+            )
+            self._raise_error(response.status, raw_data_modified)
+
+        self.log_request_success(method, str(url), url, orig_body, response.status, raw_data_modified, duration)
+
     def _tidy_url(self, url: str) -> str:
         return url.replace("%2F%7B", "/").replace("%7D%2F", "/").replace("%2F", "/")
 
@@ -656,49 +795,16 @@ async def perform_request(  # noqa: PLR0912, PLR0915
             await self._create_aiohttp_session()
         assert self.session is not None
 
-        method = _normalize_method(method)
-
-        url = self._tidy_url(url)
-        url = self._make_url_valid(url, body)
+        method, url, body, orig_body, query_string = self._prepare_async_request(method, url, params, body)
 
         if _is_root_info_request(method, url, self.base_url):
             return _mock_info_response()
         skipped_response = _skip_unsupported_endpoint(url) or _skip_multi_kb_write(self.knowledge_base, method, body)
         if skipped_response is not None:
             return skipped_response
 
-        body = _prepare_embeddings_request_body(url, method, body, self.knowledge_base)
-        orig_body = body
-        query_string = urlencode(params) if params else ""
-
-        # Top-tier tip-toeing happening here. Basically
-        # because Pip's old resolver is bad and wipes out
-        # strict pins in favor of non-strict pins of extras
-        # our [async] extra overrides aiohttp's pin of
-        # yarl. yarl released breaking changes, aiohttp pinned
-        # defensively afterwards, but our users don't get
-        # that nice pin that aiohttp set. :( So to play around
-        # this super-defensively we try to import yarl, if we can't
-        # then we pass a string into ClientSession.request() instead.
-        if query_string:
-            url = f"{url}?{query_string}"
-
         request_timeout = aiohttp.ClientTimeout(total=timeout if timeout is not None else self.timeout)
-
-        req_headers = self.headers.copy()
-        if headers:
-            req_headers.update(headers)
-
-        if self.http_compress and body:
-            body = self._gzip_compress(body)
-            req_headers["content-encoding"] = "gzip"
-
-        auth = self._http_auth if isinstance(self._http_auth, aiohttp.BasicAuth) else None
-        if callable(self._http_auth):
-            req_headers = {
-                **req_headers,
-                **self._http_auth(method, url, query_string, body),
-            }
+        req_headers, body, auth = self._build_async_request_headers(method, url, query_string, body, headers)
 
         start = self.loop.time()
         timeout_obj: ClientTimeout | None = ClientTimeout(total=timeout) if isinstance(timeout, (int, float)) else None
@@ -719,41 +825,9 @@ async def perform_request(  # noqa: PLR0912, PLR0915
         except reraise_exceptions:
             raise
         except Exception as e:  # noqa: BLE001
-            self.log_request_fail(
-                method,
-                str(url),
-                url,
-                orig_body,
-                self.loop.time() - start,
-                exception=e,
-            )
-            if isinstance(e, aiohttp_exceptions.ServerFingerprintMismatch):
-                raise SSLError("N/A", str(e), e) from e
-            if isinstance(e, (asyncio.TimeoutError, aiohttp_exceptions.ServerTimeoutError)):
-                raise ConnectionTimeout("TIMEOUT", str(e), e) from e
-            raise ConnectionError("N/A", str(e), e) from e
-
-        # raise warnings if any from the 'Warnings' header.
-        warning_headers = response.headers.getall("warning", ())
-        self._raise_warnings(warning_headers)
-
-        raw_data_modified = str(_modify_post_response_langchain(raw_data))
-
-        # raise errors based on http status codes, let the client handle those if needed
-        if not (HTTP_OK <= response.status < HTTP_MULTIPLE_CHOICES) and response.status not in ignore:
-            self.log_request_fail(
-                method,
-                str(url),
-                url,
-                orig_body,
-                duration,
-                status_code=response.status,
-                response=raw_data_modified,
-            )
-            self._raise_error(response.status, raw_data_modified)
-
-        self.log_request_success(method, str(url), url, orig_body, response.status, raw_data_modified, duration)
+            self._raise_async_request_exception(method, url, orig_body, start, e)
 
+        self._handle_async_response(method, url, orig_body, response, raw_data, duration, ignore)
         return response.status, response.headers, raw_data
 
     async def close(self) -> Any: