如何捕获Cloud SQL API返回的ClientResponseError错误？

为什么原有捕获逻辑不生效 #

你遇到的ClientResponseError(429)是在Cloud SQL Connector内部的异步后台刷新任务（_refresh_task）中抛出的，这个任务负责定期刷新实例元数据和临时证书。由于它是后台异步执行的，异常不会直接冒泡到你外层的同步try-except代码块（pool.connect()/connection.execute()所在的块），所以你的捕获逻辑根本触发不到。

正确的处理方式：在连接创建入口捕获并重试 #

需要将错误捕获和退避逻辑放到getconn函数内部——也就是connector.connect()的调用位置，因为连接池在创建/重建连接时都会调用这个函数，这里是拦截连接相关错误的最佳入口。

方案1：手动实现指数退避重试

import os
import time
import logging
from aiohttp.client_exceptions import ClientResponseError
from google.cloud.sql.connector import connector, IPTypes
import pg8000.dbapi
import sqlalchemy

ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC
logger = logging.getLogger(__name__)

def getconn() -> pg8000.dbapi.Connection:
    max_retries = 3
    base_delay = 30  # 初始退避时间
    
    for attempt in range(max_retries):
        try:
            return connector.connect(
                "myinstance",
                "pg8000",
                user="myuser",
                db="mydb",
                enable_iam_auth=True,
                ip_type=ip_type,
            )
        except ClientResponseError as ex:
            if ex.status == 429 and attempt < max_retries - 1:
                # 指数退避：每重试一次，等待时间翻倍
                backoff_seconds = base_delay * (2 ** attempt)
                logger.warning(f"Cloud SQL API触发429限流，第{attempt+1}/{max_retries}次重试，等待{backoff_seconds}秒后重试")
                time.sleep(backoff_seconds)
            else:
                logger.error(f"Cloud SQL连接失败，已重试{max_retries}次，错误详情：{ex}")
                raise  # 超出重试次数，抛出异常

# 初始化SQLAlchemy连接池
pool = sqlalchemy.create_engine(
    "postgresql+pg8000://",
    creator=getconn,
)

# 业务逻辑执行
try:
    with pool.connect() as connection:
        result = connection.execute(query).fetchall()
        for row in result:
            result_value = row[0]
except Exception as ex:
    # 处理其他业务相关异常
    logger.error(f"业务执行出错：{ex}")
    raise

方案2：用tenacity库简化重试逻辑

如果不想手动写重试逻辑，可以用tenacity库实现更优雅的指数退避：

import os
import logging
from aiohttp.client_exceptions import ClientResponseError
from google.cloud.sql.connector import connector, IPTypes
import pg8000.dbapi
import sqlalchemy
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type, retry_if_result

ip_type = IPTypes.PRIVATE if os.environ.get("PRIVATE_IP") else IPTypes.PUBLIC
logger = logging.getLogger(__name__)

# 配置重试规则：最多3次，指数退避（30秒起步，最长120秒），仅捕获429错误
@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=30, max=120),
    retry=(retry_if_exception_type(ClientResponseError) & retry_if_result(lambda e: e.status == 429)),
    before_sleep=lambda retry_state: logger.warning(f"Cloud SQL API触发429限流，第{retry_state.attempt_number}次重试，等待{retry_state.next_action.sleep}秒")
)
def getconn() -> pg8000.dbapi.Connection:
    return connector.connect(
        "myinstance",
        "pg8000",
        user="myuser",
        db="mydb",
        enable_iam_auth=True,
        ip_type=ip_type,
    )

# 初始化连接池和业务逻辑同方案1

补充说明 #

1. Cloud SQL Connector本身对后台刷新任务的429错误有基础重试机制，但如果限流频繁，在连接创建层添加重试能进一步提升稳定性。
2. 指数退避比固定等待更合理，能避免短时间内重复请求加重API服务器负担。
3. 确保正确导入aiohttp.client_exceptions.ClientResponseError，否则会出现捕获不到的情况。

内容的提问来源于stack exchange，提问作者red888