Repository: apache/airflow
File: providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py (Python)

Code before:
            ConnectionError | Timeout | ClientConnectionError | asyncio.TimeoutError,
        ):
            return True
        return False

    def _make_api_call_with_retries(
        self, method: str, url: str, headers: dict, params: dict | None = None, json: dict | None = None
    ):
        """
        Make an API call to the Snowflake SQL API with retry logic for specific HTTP errors.

        Error handling implemented based on Snowflake error handling docs:
        https://docs.snowflake.com/en/developer-guide/sql-api/handling-errors

        :param method: The HTTP method to use for the API call.
        :param url: The URL for the API endpoint.
        :param headers: The headers to include in the API call.
        :param params: (Optional) The query parameters to include in the API call.
        :param json: (Optional) The data to include in the API call.
        :return: The response object from the API call.
        """
        with requests.Session() as session:
            for attempt in Retrying(**self.retry_config):  # type: ignore
                with attempt:
                    if method.upper() in ("GET", "POST"):
                        request_kwargs = dict(method=method.lower(), url=url, headers=headers, params=params, json=json)
                        if self.http_timeout_seconds is not None:
                            request_kwargs["timeout"] = self.http_timeout_seconds
                        response = session.request(**request_kwargs)
                    else:
                        raise ValueError(f"Unsupported HTTP method: {method}")
                    response.raise_for_status()
                    return response.status_code, response.json()

    async def _make_api_call_with_retries_async(self, method, url, headers, params=None):
        """
        Make an API call to the Snowflake SQL API asynchronously with retry logic for specific HTTP errors.

        Error handling implemented based on Snowflake error handling docs:
        https://docs.snowflake.com/en/developer-guide/sql-api/handling-errors

        :param method: The HTTP method to use for the API call. Only GET is supported as  is synchronous.
        :param url: The URL for the API endpoint.
        :param headers: The headers to include in the API call.
        :param params: (Optional) The query parameters to include in the API call.
        :return: The response object from the API call.
        """
        # aiohttp typically applies a default total timeout if not specified.
        # We override it when `http_timeout_seconds` is provided.
        # For simplicity, we only configure the total timeout (no per-phase tuning).
        # See https://docs.aiohttp.org/en/stable/client_reference.html#aiohttp.ClientTimeout

Review comment (suggestion):
I'll make this change with the below review is complete.
https://github.com/apache/airflow/pull/60689/files#r2700868724
```python
request_kwargs: dict[str, Any] = {
    "method": method.lower(),
    "url": url,
    "headers": headers,
    "params": params,
    "json": json,
}
```

Code after:

    @staticmethod
    def _should_retry_on_error(exception) -> bool:
        """
        Determine if the exception should trigger a retry based on error type and status code.

        Retries on HTTP errors 429 (Too Many Requests), 503 (Service Unavailable),
        and 504 (Gateway Timeout) as recommended by Snowflake error handling docs.
        Retries on connection errors and timeouts.

        :param exception: The exception to check
        :return: True if the request should be retried, False otherwise
        """
        if isinstance(exception, HTTPError):
            return exception.response.status_code in [429, 503, 504]
        if isinstance(exception, ClientResponseError):
            return exception.status in [429, 503, 504]
        if isinstance(
            exception,
            ConnectionError | Timeout | ClientConnectionError | asyncio.TimeoutError,
        ):
            return True
        return False

    def _make_api_call_with_retries(
        self, method: str, url: str, headers: dict, params: dict | None = None, json: dict | None = None
    ):
        """
        Make an API call to the Snowflake SQL API with retry logic for specific HTTP errors.

        Error handling implemented based on Snowflake error handling docs:
        https://docs.snowflake.com/en/developer-guide/sql-api/handling-errors

        :param method: The HTTP method to use for the API call.
        :param url: The URL for the API endpoint.
        :param headers: The headers to include in the API call.
        :param params: (Optional) The query parameters to include in the API call.
        :param json: (Optional) The data to include in the API call.
        :return: The response object from the API call.
        """
        if method.upper() not in ("GET", "POST"):
            raise ValueError(f"Unsupported HTTP method: {method}")

        user_kwargs: dict[str, Any] = dict(self.http_request_kwargs or {})
        forbidden: set[str] = self.HTTP_REQUEST_KWARGS_GUARD_KEYS & set(user_kwargs)
        if forbidden:
            raise ValueError(
                f"http_request_kwargs must not override request identity fields: {sorted(forbidden)}"
            )

        with requests.Session() as session: