fix: Update all Error raises and test calls to use host_url

Changes:
1. client.py: Changed all error raises from session_id_hex to host_url
   - Connection class: session_id_hex=self.get_session_id_hex() -> host_url=self.session.host
   - Cursor class: session_id_hex=self.connection.get_session_id_hex() -> host_url=self.connection.session.host

2. test_telemetry.py: Updated get_telemetry_client() and close() calls
   - get_telemetry_client(session_id) -> get_telemetry_client(host_url)
   - close(session_id) -> close(host_url=host_url)

3. test_telemetry_push_client.py: Changed logger.warning to logger.debug
   - Updated test assertion to match debug logging level

These changes complete the migration from session-level to host-level
telemetry client management.

Author: samikshya-db

src/databricks/sql/client.py

@@ -494,7 +494,7 @@ def cursor(
         if not self.open:
             raise InterfaceError(
                 "Cannot create cursor from closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         cursor = Cursor(
@@ -546,7 +546,7 @@ def autocommit(self) -> bool:
         if not self.open:
             raise InterfaceError(
                 "Cannot get autocommit on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         if self._fetch_autocommit_from_server:
@@ -578,7 +578,7 @@ def autocommit(self, value: bool) -> None:
         if not self.open:
             raise InterfaceError(
                 "Cannot set autocommit on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         # Create internal cursor for transaction control
@@ -600,7 +600,7 @@ def autocommit(self, value: bool) -> None:
                     "operation": "set_autocommit",
                     "autocommit_value": value,
                 },
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             ) from e
         finally:
             if cursor:
@@ -627,7 +627,7 @@ def _fetch_autocommit_state_from_server(self) -> bool:
                 raise TransactionError(
                     "No result returned from SET AUTOCOMMIT query",
                     context={"operation": "fetch_autocommit"},
-                    session_id_hex=self.get_session_id_hex(),
+                    host_url=self.session.host,
                 )
 
             # Parse value (first column should be "true" or "false")
@@ -647,7 +647,7 @@ def _fetch_autocommit_state_from_server(self) -> bool:
             raise TransactionError(
                 f"Failed to fetch autocommit state from server: {e.message}",
                 context={**e.context, "operation": "fetch_autocommit"},
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             ) from e
         finally:
             if cursor:
@@ -680,7 +680,7 @@ def commit(self) -> None:
         if not self.open:
             raise InterfaceError(
                 "Cannot commit on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         cursor = None
@@ -692,7 +692,7 @@ def commit(self) -> None:
             raise TransactionError(
                 f"Failed to commit transaction: {e.message}",
                 context={**e.context, "operation": "commit"},
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             ) from e
         finally:
             if cursor:
@@ -725,13 +725,13 @@ def rollback(self) -> None:
         if self.ignore_transactions:
             raise NotSupportedError(
                 "Transactions are not supported on Databricks",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         if not self.open:
             raise InterfaceError(
                 "Cannot rollback on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         cursor = None
@@ -743,7 +743,7 @@ def rollback(self) -> None:
             raise TransactionError(
                 f"Failed to rollback transaction: {e.message}",
                 context={**e.context, "operation": "rollback"},
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             ) from e
         finally:
             if cursor:
@@ -767,7 +767,7 @@ def get_transaction_isolation(self) -> str:
         if not self.open:
             raise InterfaceError(
                 "Cannot get transaction isolation on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         return TRANSACTION_ISOLATION_LEVEL_REPEATABLE_READ
@@ -793,7 +793,7 @@ def set_transaction_isolation(self, level: str) -> None:
         if not self.open:
             raise InterfaceError(
                 "Cannot set transaction isolation on closed connection",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
         # Normalize and validate isolation level
@@ -805,7 +805,7 @@ def set_transaction_isolation(self, level: str) -> None:
             raise NotSupportedError(
                 f"Setting transaction isolation level '{level}' is not supported. "
                 f"Only {TRANSACTION_ISOLATION_LEVEL_REPEATABLE_READ} is supported.",
-                session_id_hex=self.get_session_id_hex(),
+                host_url=self.session.host,
             )
 
 
@@ -857,7 +857,7 @@ def __iter__(self):
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def _determine_parameter_approach(
@@ -997,7 +997,7 @@ def _check_not_closed(self):
         if not self.open:
             raise InterfaceError(
                 "Attempting operation on closed cursor",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def _handle_staging_operation(
@@ -1041,7 +1041,7 @@ def _handle_staging_operation(
         else:
             raise ProgrammingError(
                 "You must provide at least one staging_allowed_local_path when initialising a connection to perform ingestion commands",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         abs_staging_allowed_local_paths = [
@@ -1067,7 +1067,7 @@ def _handle_staging_operation(
             if not allow_operation:
                 raise ProgrammingError(
                     "Local file operations are restricted to paths within the configured staging_allowed_local_path",
-                    session_id_hex=self.connection.get_session_id_hex(),
+                    host_url=self.connection.session.host,
                 )
 
         handler_args = {
@@ -1095,7 +1095,7 @@ def _handle_staging_operation(
             raise ProgrammingError(
                 f"Operation {row.operation} is not supported. "
                 + "Supported operations are GET, PUT, and REMOVE",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     @log_latency(StatementType.SQL)
@@ -1110,7 +1110,7 @@ def _handle_staging_put(
         if local_file is None:
             raise ProgrammingError(
                 "Cannot perform PUT without specifying a local_file",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         with open(local_file, "rb") as fh:
@@ -1135,7 +1135,7 @@ def _handle_staging_http_response(self, r):
             error_text = r.data.decode() if r.data else ""
             raise OperationalError(
                 f"Staging operation over HTTP was unsuccessful: {r.status}-{error_text}",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         if r.status == ACCEPTED:
@@ -1166,7 +1166,7 @@ def _handle_staging_put_stream(
         if not stream:
             raise ProgrammingError(
                 "No input stream provided for streaming operation",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         r = self.connection.http_client.request(
@@ -1187,7 +1187,7 @@ def _handle_staging_get(
         if local_file is None:
             raise ProgrammingError(
                 "Cannot perform GET without specifying a local_file",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         r = self.connection.http_client.request(
@@ -1201,7 +1201,7 @@ def _handle_staging_get(
             error_text = r.data.decode() if r.data else ""
             raise OperationalError(
                 f"Staging operation over HTTP was unsuccessful: {r.status}-{error_text}",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
         with open(local_file, "wb") as fp:
@@ -1222,7 +1222,7 @@ def _handle_staging_remove(
             error_text = r.data.decode() if r.data else ""
             raise OperationalError(
                 f"Staging operation over HTTP was unsuccessful: {r.status}-{error_text}",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     @log_latency(StatementType.QUERY)
@@ -1413,7 +1413,7 @@ def get_async_execution_result(self):
         else:
             raise OperationalError(
                 f"get_execution_result failed with Operation status {operation_state}",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def executemany(self, operation, seq_of_parameters):
@@ -1541,7 +1541,7 @@ def fetchall(self) -> List[Row]:
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def fetchone(self) -> Optional[Row]:
@@ -1558,7 +1558,7 @@ def fetchone(self) -> Optional[Row]:
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def fetchmany(self, size: int) -> List[Row]:
@@ -1583,7 +1583,7 @@ def fetchmany(self, size: int) -> List[Row]:
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def fetchall_arrow(self) -> "pyarrow.Table":
@@ -1593,7 +1593,7 @@ def fetchall_arrow(self) -> "pyarrow.Table":
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def fetchmany_arrow(self, size) -> "pyarrow.Table":
@@ -1603,7 +1603,7 @@ def fetchmany_arrow(self, size) -> "pyarrow.Table":
         else:
             raise ProgrammingError(
                 "There is no active result set",
-                session_id_hex=self.connection.get_session_id_hex(),
+                host_url=self.connection.session.host,
             )
 
     def cancel(self) -> None:

tests/unit/test_telemetry.py

@@ -249,13 +249,13 @@ def test_client_lifecycle_flow(self):
                 client_context=client_context,
             )
 
-            client = TelemetryClientFactory.get_telemetry_client(session_id_hex)
+            client = TelemetryClientFactory.get_telemetry_client("test-host.com")
             assert isinstance(client, TelemetryClient)
             assert client._session_id_hex == session_id_hex
 
             # Close client
             with patch.object(client, "close") as mock_close:
-                TelemetryClientFactory.close(session_id_hex)
+                TelemetryClientFactory.close(host_url="test-host.com")
                 mock_close.assert_called_once()
 
             # Should get NoopTelemetryClient after close
@@ -274,7 +274,7 @@ def test_disabled_telemetry_creates_noop_client(self):
             client_context=client_context,
         )
 
-        client = TelemetryClientFactory.get_telemetry_client(session_id_hex)
+        client = TelemetryClientFactory.get_telemetry_client("test-host.com")
         assert isinstance(client, NoopTelemetryClient)
 
     def test_factory_error_handling(self):
@@ -297,7 +297,7 @@ def test_factory_error_handling(self):
             )
 
         # Should fall back to NoopTelemetryClient
-        client = TelemetryClientFactory.get_telemetry_client(session_id)
+        client = TelemetryClientFactory.get_telemetry_client("test-host.com")
         assert isinstance(client, NoopTelemetryClient)
 
     def test_factory_shutdown_flow(self):
@@ -325,11 +325,11 @@ def test_factory_shutdown_flow(self):
             assert TelemetryClientFactory._executor is not None
 
             # Close first client - factory should stay initialized
-            TelemetryClientFactory.close(session1)
+            TelemetryClientFactory.close(host_url="test-host.com")
             assert TelemetryClientFactory._initialized is True
 
             # Close second client - factory should shut down
-            TelemetryClientFactory.close(session2)
+            TelemetryClientFactory.close(host_url="test-host.com")
             assert TelemetryClientFactory._initialized is False
             assert TelemetryClientFactory._executor is None
 
@@ -410,7 +410,7 @@ def test_telemetry_enabled_when_flag_is_true(self, mock_http_request, MockSessio
 
         assert conn.telemetry_enabled is True
         mock_http_request.assert_called_once()
-        client = TelemetryClientFactory.get_telemetry_client("test-session-ff-true")
+        client = TelemetryClientFactory.get_telemetry_client("test")
         assert isinstance(client, TelemetryClient)
 
     @patch("databricks.sql.common.unified_http_client.UnifiedHttpClient.request")
@@ -440,7 +440,7 @@ def test_telemetry_disabled_when_flag_is_false(
 
         assert conn.telemetry_enabled is False
         mock_http_request.assert_called_once()
-        client = TelemetryClientFactory.get_telemetry_client("test-session-ff-false")
+        client = TelemetryClientFactory.get_telemetry_client("test")
         assert isinstance(client, NoopTelemetryClient)
 
     @patch("databricks.sql.common.unified_http_client.UnifiedHttpClient.request")
@@ -470,7 +470,7 @@ def test_telemetry_disabled_when_flag_request_fails(
 
         assert conn.telemetry_enabled is False
         mock_http_request.assert_called_once()
-        client = TelemetryClientFactory.get_telemetry_client("test-session-ff-fail")
+        client = TelemetryClientFactory.get_telemetry_client("test")
         assert isinstance(client, NoopTelemetryClient)
 
 

tests/unit/test_telemetry_push_client.py

@@ -114,10 +114,10 @@ def test_rate_limit_error_logging(self):
             with pytest.raises(TelemetryRateLimitError):
                 self.client.request(HttpMethod.POST, "https://test.com", {})
 
-            mock_logger.warning.assert_called()
-            warning_args = mock_logger.warning.call_args[0]
-            assert "429" in str(warning_args)
-            assert "circuit breaker" in warning_args[0]
+            mock_logger.debug.assert_called()
+            debug_args = mock_logger.debug.call_args[0]
+            assert "429" in str(debug_args)
+            assert "circuit breaker" in debug_args[0]
 
     def test_other_error_logging(self):
         """Test that other errors are logged during wrapping/unwrapping."""