解决Cloud SQL连接中的KeyError：Python连接器参数传递指南

instance_connection_name = os.environ[self.keys["gProj"]] # 预期是 'project:region:instance'
db_user = os.environ[self.keys["gUser"]]
db_pass = os.environ[self.keys["gPass"]]
db_name = os.environ[self.keys["gDB"]]

import os
import sqlalchemy
import pg8000
from google.cloud.sql.connector import Connector, IPTypes

class CloudSQLConnector:
    def __init__(self, keys: dict):
        self.keys = keys # 假设self.keys包含了所有必要的连接信息

    def connect_with_connector(self) -> sqlalchemy.engine.base.Engine:
        """
        初始化一个Postgres Cloud SQL实例的连接池。
        使用Cloud SQL Python连接器包。
        """
        # 注意: 将凭证保存在环境变量中虽然方便，但不够安全。
        # 考虑使用更安全的解决方案，如Cloud Secret Manager，以帮助保护密钥。

        # 根据环境变量判断IP类型，如果没有设置则默认为公共IP
        ip_type = IPTypes.PRIVATE if os.environ.get(self.keys.get("gPrivIP_ENV_KEY")) else IPTypes.PUBLIC

        # 初始化Cloud SQL Python连接器对象
        connector = Connector()

        def getconn() -> pg8000.dbapi.Connection:
            # 直接从self.keys获取参数，避免os.environ的KeyError
            conn: pg8000.dbapi.Connection = connector.connect(
                self.keys["gProj"],  # 实例连接名称，例如 'project:region:instance'
                "pg8000",             # 数据库驱动名称
                user=self.keys["gUser"],
                password=self.keys["gPass"],
                db=self.keys["gDB"],
                ip_type=ip_type,
            )
            return conn

        # Cloud SQL Python连接器可以与SQLAlchemy结合使用
        # 通过 'creator' 参数传递 getconn 函数
        pool = sqlalchemy.create_engine(
            "postgresql+pg8000://",
            creator=getconn,
            # 其他SQLAlchemy连接池参数可以按需添加，例如：
            # pool_size=5,
            # max_overflow=2,
            # pool_timeout=30, # seconds
            # pool_recycle=1800 # seconds
        )
        return pool

# 示例用法 (假设您已经有了配置字典)
# config_keys = {
#     "gProj": "your-project-id:your-region:your-instance-name",
#     "gUser": "your-db-user",
#     "gPass": "your-db-password",
#     "gDB": "your-database-name",
#     "gPrivIP_ENV_KEY": "USE_PRIVATE_IP" # 这是一个示例环境变量键，用于控制IP类型
# }
# connector_instance = CloudSQLConnector(config_keys)
# db_pool = connector_instance.connect_with_connector()
# with db_pool.connect() as conn:
#     result = conn.execute(sqlalchemy.text("SELECT 1")).scalar()
#     print(f"Connection successful: {result}")