Repository: GoogleCloudPlatform/cloud-sql-python-connector (python)

Issue:
Improve `aiohttp` error handling by converting response body to error
The Python Connector currently uses `raise_for_status=True` when using aiohttp's [`ClientSession`](https://docs.aiohttp.org/en/stable/client_reference.html). 

The default behaviour for this argument is to close http connections when response status code > 400 and raise generic error message (ex. `400 Bad Request`), so the response body which may have a more helpful error message or instructions cannot be accessed.

It is recommended to not use `raise for status` if access to the errors response body is required. Because of this the Python Connector should do something along the following (pseudocode):

```python

def convert_response(resp):
  # attempt to read response and format custom error
  try: 
    resp_body = await resp.json()
    # ... raise error using standard response format
    build_error(resp.status, resp_body)
  # if error occurs reading response body, fallback to raise_for_status 
  except:
    resp.raise_for_status()

# --------------------------------------------------------------------
# ... inside _get_metadata (refresh_utils.py)
async with client_session.get(url, headers=headers) as resp:
  # if not 2XX status code
  if not resp.ok:
    # handle standard response and format it into error
    convert_response(resp)
  # ... continue onwards
```    


Pull request description:
feat: improve aiohttp client error messages
By default, `aiohttp` on bad requests will discard response body and
output generic error message in response, `Bad Request`, `Forbidden`, etc.

The Cloud SQL Admin APIs response body contains more detailed error messages.
We need to raise these to the end user for them to be able to resolve common config 
issues on their own.

In https://github.com/GoogleCloudPlatform/cloud-sql-python-connector/pull/362 we caught `403` errors and implemented a custom error message
to point users at the most common config issues, Cloud SQL API not being enabled,
missing `Cloud SQL Client` role etc. However, this was not a catch-all (as noted by #1199) 
and more of a temporary fix.

This PR implements a more robust solution, copying the actual Cloud SQL Admin API
response body's error message to the end user.

> [!NOTE]
> Example common scenario: Calling the Python Connector with a typo in your instance name.

Error Message today:

```
aiohttp.client_exceptions.ClientResponseError: 404, message='Not Found'
```

Error Message after PR:

```
aiohttp.client_exceptions.ClientResponseError: 404, message='The Cloud SQL instance does not exist.'
```

Related https://github.com/aio-libs/aiohttp/issues/4600
I looked into using a custom response class but it seemed over the top
compared to the working approach here.

Closes #477 

Patch:
diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml
index ba64398..1116c99 100644
--- a/.github/workflows/codeql.yml
+++ b/.github/workflows/codeql.yml
@@ -46,16 +46,16 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@ea9e4e37992a54ee68a9622e985e60c8e8f12d9f # v3.27.4
+        uses: github/codeql-action/init@f09c1c0a94de965c15400f5634aa42fac8fb8f88 # v3.27.5
         with:
           languages: ${{ matrix.language }}
 
       # Autobuild attempts to build any compiled languages  (C/C++, C#, Go, or Java).
       # If this step fails, then you should remove it and run the build manually
       - name: Autobuild
-        uses: github/codeql-action/autobuild@ea9e4e37992a54ee68a9622e985e60c8e8f12d9f # v3.27.4
+        uses: github/codeql-action/autobuild@f09c1c0a94de965c15400f5634aa42fac8fb8f88 # v3.27.5
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@ea9e4e37992a54ee68a9622e985e60c8e8f12d9f # v3.27.4
+        uses: github/codeql-action/analyze@f09c1c0a94de965c15400f5634aa42fac8fb8f88 # v3.27.5
         with:
           category: "/language:${{matrix.language}}"
diff --git a/.github/workflows/scorecard.yml b/.github/workflows/scorecard.yml
index 37c9428..ee3efa1 100644
--- a/.github/workflows/scorecard.yml
+++ b/.github/workflows/scorecard.yml
@@ -65,6 +65,6 @@ jobs:
 
       # Upload the results to GitHub's code scanning dashboard.
       - name: "Upload to code-scanning"
-        uses: github/codeql-action/upload-sarif@ea9e4e37992a54ee68a9622e985e60c8e8f12d9f # v3.27.4
+        uses: github/codeql-action/upload-sarif@f09c1c0a94de965c15400f5634aa42fac8fb8f88 # v3.27.5
         with:
           sarif_file: resultsFiltered.sarif
diff --git a/google/cloud/sql/connector/client.py b/google/cloud/sql/connector/client.py
index ed305ec..61b77d5 100644
--- a/google/cloud/sql/connector/client.py
+++ b/google/cloud/sql/connector/client.py
@@ -128,8 +128,19 @@ class CloudSQLClient:
         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 @@ class CloudSQLClient:
         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"]
 
diff --git a/google/cloud/sql/connector/instance.py b/google/cloud/sql/connector/instance.py
index f244b8c..9cf9bc7 100644
--- a/google/cloud/sql/connector/instance.py
+++ b/google/cloud/sql/connector/instance.py
@@ -22,8 +22,6 @@ from datetime import timedelta
 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 @@ class RefreshAheadCache:
                 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 "
diff --git a/requirements.txt b/requirements.txt
index e285d4a..466849d 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,5 @@
 aiofiles==24.1.0
-aiohttp==3.11.7
-cryptography==43.0.3
+aiohttp==3.11.9
+cryptography==44.0.0
 Requests==2.32.3
 google-auth==2.36.0


Test patch:
diff --git a/requirements-test.txt b/requirements-test.txt
index 4aeeced..40102fc 100644
--- a/requirements-test.txt
+++ b/requirements-test.txt
@@ -1,4 +1,4 @@
-pytest==8.3.3
+pytest==8.3.4
 mock==5.1.0
 pytest-cov==6.0.0
 pytest-asyncio==0.24.0
diff --git a/tests/unit/test_client.py b/tests/unit/test_client.py
index 046e8e5..af42af0 100644
--- a/tests/unit/test_client.py
+++ b/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,130 @@ 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_get_metadata_error_parsing_json(
+    fake_credentials: Credentials,
+) -> None:
+    """
+    Test that aiohttp default error messages are raised when _get_metadata gets
+    a bad JSON response.
+    """
+    # 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"]  # invalid JSON
+    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 == "Forbidden"
+        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()
+
+
+async def test_get_ephemeral_error_parsing_json(
+    fake_credentials: Credentials,
+) -> None:
+    """
+    Test that aiohttp default error messages are raised when _get_ephemeral gets
+    a bad JSON response.
+    """
+    # 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"]  # invalid JSON
+    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 == "Not Found"
+        await client.close()
diff --git a/tests/unit/test_instance.py b/tests/unit/test_instance.py
index 5b0887a..3ce0386 100644
--- a/tests/unit/test_instance.py
+++ b/tests/unit/test_instance.py
@@ -17,10 +17,6 @@ limitations under the License.
 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:
     """