Merge SeaHttpClient into THttpClient

src/databricks/sql/auth/retry.py

@@ -127,7 +127,7 @@ def __init__(
             total=_attempts_remaining,
             respect_retry_after_header=True,
             backoff_factor=self.delay_min,
-            allowed_methods=["POST"],
+            allowed_methods=["POST", "DELETE", "GET"],
             status_forcelist=[429, 503, *self.force_dangerous_codes],
         )
 

src/databricks/sql/auth/thrift_http_client.py

@@ -1,16 +1,18 @@
 import base64
+import json
 import logging
 import urllib.parse
-from typing import Dict, Union, Optional
+from typing import Dict, Union, Optional, Any
 
 import six
-import thrift
+import thrift.transport.THttpClient
 
 import ssl
 import warnings
 from http.client import HTTPResponse
 from io import BytesIO
 
+import urllib3
 from urllib3 import HTTPConnectionPool, HTTPSConnectionPool, ProxyManager
 from urllib3.util import make_headers
 from databricks.sql.auth.retry import CommandType, DatabricksRetryPolicy
@@ -222,3 +224,105 @@ def set_retry_command_type(self, value: CommandType):
             logger.warning(
                 "DatabricksRetryPolicy is currently bypassed. The CommandType cannot be set."
             )
+
+    def make_rest_request(
+        self,
+        method: str,
+        endpoint_path: str,
+        data: Optional[Dict[str, Any]] = None,
+        params: Optional[Dict[str, Any]] = None,
+        headers: Optional[Dict[str, str]] = None,
+    ) -> Dict[str, Any]:
+        """
+        Make a REST API request using the existing connection pool.
+
+        Args:
+            method (str): HTTP method (GET, POST, DELETE, etc.)
+            endpoint_path (str): API endpoint path (e.g., "sessions" or "statements/123")
+            data (dict, optional): Request payload data
+            params (dict, optional): Query parameters
+            headers (dict, optional): Additional headers
+
+        Returns:
+            dict: Response data parsed from JSON
+
+        Raises:
+            RequestError: If the request fails
+        """
+        # Ensure the transport is open
+        if not self.isOpen():
+            self.open()
+
+        # Prepare headers
+        request_headers = {
+            "Content-Type": "application/json",
+        }
+
+        # Add authentication headers
+        auth_headers: Dict[str, str] = {}
+        self.__auth_provider.add_headers(auth_headers)
+        request_headers.update(auth_headers)
+
+        # Add custom headers if provided
+        if headers:
+            request_headers.update(headers)
+
+        # Prepare request body
+        body = json.dumps(data).encode("utf-8") if data else None
+
+        # Build query string for params
+        query_string = ""
+        if params:
+            query_string = "?" + urllib.parse.urlencode(params)
+
+        # Determine full path
+        full_path = (
+            self.path.rstrip("/") + "/" + endpoint_path.lstrip("/") + query_string
+        )
+
+        # Log request details (debug level)
+        logger.debug(f"Making {method} request to {full_path}")
+
+        # Make request using the connection pool - let urllib3 exceptions propagate
+        logger.debug(f"making request to {full_path}")
+        logger.debug(f"\trequest headers: {request_headers}")
+        logger.debug(f"\trequest body: {body.decode('utf-8') if body else None}")
+        logger.debug(f"\trequest params: {params}")
+        logger.debug(f"\trequest full path: {full_path}")
+        self.__resp = self.__pool.request(
+            method,
+            url=full_path,
+            body=body,
+            headers=request_headers,
+            preload_content=False,
+            timeout=self.__timeout,
+            retries=self.retry_policy,
+        )
+        logger.debug(f"Response: {self.__resp}")
+
+        # Store response status and headers
+        if self.__resp is not None:
+            self.code = self.__resp.status
+            self.message = self.__resp.reason
+            self.headers = self.__resp.headers
+
+            # Log response status
+            logger.debug(f"Response status: {self.code}, message: {self.message}")
+
+            # Read and parse response data
+            # Note: urllib3's HTTPResponse has a data attribute, but it's not in the type stubs
+            response_data = getattr(self.__resp, "data", None)
+
+            # Parse JSON response if there is content
+            if response_data:
+                result = json.loads(response_data.decode("utf-8"))
+
+                # Log response content (truncated for large responses)
+                content_str = json.dumps(result)
+                logger.debug(f"Response content: {content_str}")
+
+                return result
+
+            return {}
+        else:
+            raise ValueError("No response received from server")

src/databricks/sql/backend/filters.py

@@ -13,10 +13,12 @@
     cast,
 )
 
-from databricks.sql.backend.sea.backend import SeaDatabricksClient
-from databricks.sql.backend.types import ExecuteResponse
+from databricks.sql.backend.types import ExecuteResponse, CommandId
+from databricks.sql.backend.sea.models.base import ResultData
 
-from databricks.sql.result_set import ResultSet, SeaResultSet
+if TYPE_CHECKING:
+    from databricks.sql.backend.sea.backend import SeaDatabricksClient
+    from databricks.sql.result_set import ResultSet, SeaResultSet
 
 logger = logging.getLogger(__name__)
 
@@ -70,6 +72,8 @@ def _filter_sea_result_set(
         from databricks.sql.result_set import SeaResultSet
 
         # Create a new SeaResultSet with the filtered data
+        from databricks.sql.backend.sea.backend import SeaDatabricksClient
+
         filtered_result_set = SeaResultSet(
             connection=result_set.connection,
             execute_response=execute_response,