feat: improve aiohttp client error messages

google/cloud/sql/connector/client.py

@@ -128,8 +128,19 @@ async def _get_metadata(
         resp = await self._client.get(url, headers=headers)
         if resp.status >= 500:
             resp = await retry_50x(self._client.get, url, headers=headers)
-        resp.raise_for_status()
-        ret_dict = await resp.json()
+        # try to get response json for better error message
+        try:
+            ret_dict = await resp.json()
+            if resp.status >= 400:
+                # if detailed error message is in json response, use as error message
+                message = ret_dict.get("error", {}).get("message")
+                if message:
+                    resp.reason = message
+        # skip, raise_for_status will catch all errors in finally block
+        except Exception:
+            pass
+        finally:
+            resp.raise_for_status()
 
         if ret_dict["region"] != region:
             raise ValueError(
@@ -198,8 +209,19 @@ async def _get_ephemeral(
         resp = await self._client.post(url, headers=headers, json=data)
         if resp.status >= 500:
             resp = await retry_50x(self._client.post, url, headers=headers, json=data)
-        resp.raise_for_status()
-        ret_dict = await resp.json()
+        # try to get response json for better error message
+        try:
+            ret_dict = await resp.json()
+            if resp.status >= 400:
+                # if detailed error message is in json response, use as error message
+                message = ret_dict.get("error", {}).get("message")
+                if message:
+                    resp.reason = message
+        # skip, raise_for_status will catch all errors in finally block
+        except Exception:
+            pass
+        finally:
+            resp.raise_for_status()
 
         ephemeral_cert: str = ret_dict["ephemeralCert"]["cert"]
 

google/cloud/sql/connector/instance.py

@@ -22,8 +22,6 @@
 from datetime import timezone
 import logging
 
-import aiohttp
-
 from google.cloud.sql.connector.client import CloudSQLClient
 from google.cloud.sql.connector.connection_info import ConnectionInfo
 from google.cloud.sql.connector.connection_name import _parse_instance_connection_name
@@ -128,15 +126,6 @@ async def _perform_refresh(self) -> ConnectionInfo:
                 f"expiration = {connection_info.expiration.isoformat()}"
             )
 
-        except aiohttp.ClientResponseError as e:
-            logger.debug(
-                f"['{self._conn_name}']: Connection info "
-                f"refresh operation failed: {str(e)}"
-            )
-            if e.status == 403:
-                e.message = "Forbidden: Authenticated IAM principal does not seem authorized to make API request. Verify 'Cloud SQL Admin API' is enabled within your GCP project and 'Cloud SQL Client' role has been granted to IAM principal."
-            raise
-
         except Exception as e:
             logger.debug(
                 f"['{self._conn_name}']: Connection info "

tests/unit/test_client.py

@@ -15,6 +15,9 @@
 import datetime
 from typing import Optional
 
+from aiohttp import ClientResponseError
+from aioresponses import aioresponses
+from google.auth.credentials import Credentials
 from mocks import FakeCredentials
 import pytest
 
@@ -138,3 +141,72 @@ async def test_CloudSQLClient_user_agent(
         assert client._user_agent == f"cloud-sql-python-connector/{version}+{driver}"
     # close client
     await client.close()
+
+
+async def test_cloud_sql_error_messages_get_metadata(
+    fake_credentials: Credentials,
+) -> None:
+    """
+    Test that Cloud SQL Admin API error messages are raised for _get_metadata.
+    """
+    # mock Cloud SQL Admin API calls with exceptions
+    client = CloudSQLClient(
+        sqladmin_api_endpoint="https://sqladmin.googleapis.com",
+        quota_project=None,
+        credentials=fake_credentials,
+    )
+    get_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance/connectSettings"
+    resp_body = {
+        "error": {
+            "code": 403,
+            "message": "Cloud SQL Admin API has not been used in project 123456789 before or it is disabled",
+        }
+    }
+    with aioresponses() as mocked:
+        mocked.get(
+            get_url,
+            status=403,
+            payload=resp_body,
+            repeat=True,
+        )
+        with pytest.raises(ClientResponseError) as exc_info:
+            await client._get_metadata("my-project", "my-region", "my-instance")
+        assert exc_info.value.status == 403
+        assert (
+            exc_info.value.message
+            == "Cloud SQL Admin API has not been used in project 123456789 before or it is disabled"
+        )
+        await client.close()
+
+
+async def test_cloud_sql_error_messages_get_ephemeral(
+    fake_credentials: Credentials,
+) -> None:
+    """
+    Test that Cloud SQL Admin API error messages are raised for _get_ephemeral.
+    """
+    # mock Cloud SQL Admin API calls with exceptions
+    client = CloudSQLClient(
+        sqladmin_api_endpoint="https://sqladmin.googleapis.com",
+        quota_project=None,
+        credentials=fake_credentials,
+    )
+    post_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert"
+    resp_body = {
+        "error": {
+            "code": 404,
+            "message": "The Cloud SQL instance does not exist.",
+        }
+    }
+    with aioresponses() as mocked:
+        mocked.post(
+            post_url,
+            status=404,
+            payload=resp_body,
+            repeat=True,
+        )
+        with pytest.raises(ClientResponseError) as exc_info:
+            await client._get_ephemeral("my-project", "my-instance", "my-key")
+        assert exc_info.value.status == 404
+        assert exc_info.value.message == "The Cloud SQL instance does not exist."
+        await client.close()

tests/unit/test_instance.py

@@ -17,10 +17,6 @@
 import asyncio
 import datetime
 
-from aiohttp import ClientResponseError
-from aiohttp import RequestInfo
-from aioresponses import aioresponses
-from google.auth.credentials import Credentials
 from mock import patch
 import mocks
 import pytest  # noqa F401 Needed to run the tests
@@ -266,59 +262,6 @@ async def test_get_preferred_ip_CloudSQLIPTypeError(cache: RefreshAheadCache) ->
         instance_metadata.get_preferred_ip(IPTypes.PSC)
 
 
-@pytest.mark.asyncio
-async def test_ClientResponseError(
-    fake_credentials: Credentials,
-) -> None:
-    """
-    Test that detailed error message is applied to ClientResponseError.
-    """
-    # mock Cloud SQL Admin API calls with exceptions
-    keys = asyncio.create_task(generate_keys())
-    client = CloudSQLClient(
-        sqladmin_api_endpoint="https://sqladmin.googleapis.com",
-        quota_project=None,
-        credentials=fake_credentials,
-    )
-    get_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance/connectSettings"
-    post_url = "https://sqladmin.googleapis.com/sql/v1beta4/projects/my-project/instances/my-instance:generateEphemeralCert"
-    with aioresponses() as mocked:
-        mocked.get(
-            get_url,
-            status=403,
-            exception=ClientResponseError(
-                RequestInfo(get_url, "GET", headers=[]), history=[], status=403  # type: ignore
-            ),
-            repeat=True,
-        )
-        mocked.post(
-            post_url,
-            status=403,
-            exception=ClientResponseError(
-                RequestInfo(post_url, "POST", headers=[]), history=[], status=403  # type: ignore
-            ),
-            repeat=True,
-        )
-        cache = RefreshAheadCache(
-            "my-project:my-region:my-instance",
-            client,
-            keys,
-        )
-        try:
-            await cache._current
-        except ClientResponseError as e:
-            assert e.status == 403
-            assert (
-                e.message == "Forbidden: Authenticated IAM principal does not "
-                "seem authorized to make API request. Verify "
-                "'Cloud SQL Admin API' is enabled within your GCP project and "
-                "'Cloud SQL Client' role has been granted to IAM principal."
-            )
-        finally:
-            await cache.close()
-            await client.close()
-
-
 @pytest.mark.asyncio
 async def test_AutoIAMAuthNotSupportedError(fake_client: CloudSQLClient) -> None:
     """