解决Cloud SQL Python连接器KeyError：环境变量与参数直传

instance_connection_name = os.environ[self.keys["gProj"]]

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

class DatabaseConnector:
    def __init__(self, keys_config):
        # 假设 self.keys_config 包含了所有必要的连接参数值
        # 例如: keys_config = {
        #     "gProj": "your-project-id:your-region:your-instance-name",
        #     "gUser": "your-db-user",
        #     "gPass": "your-db-password",
        #     "gDB": "your-database-name",
        #     "gPrivIP": "SOME_ENV_VAR_FOR_PRIVATE_IP_CHECK" # 这是一个环境变量的名称
        # }
        self.keys = keys_config

    def connect_with_connector(self) -> sqlalchemy.engine.base.Engine:
        """
        初始化一个Postgres Cloud SQL实例的连接池。

        使用Cloud SQL Python Connector包。
        """
        # 注意: 将凭据保存在环境变量中虽然方便，但不够安全。
        # 考虑使用更安全的解决方案，如Cloud Secret Manager (https://cloud.google.com/secret-manager) 来帮助保护秘密信息。

        # 根据环境变量是否存在来决定使用私有IP还是公共IP
        # 这里假设 self.keys["gPrivIP"] 存储的是一个环境变量的名称，用于检查私有IP配置
        ip_type = IPTypes.PRIVATE if os.environ.get(self.keys["gPrivIP"]) else IPTypes.PUBLIC

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

        def getconn() -> pg8000.dbapi.Connection:
            conn: pg8000.dbapi.Connection = connector.connect(
                # 直接使用 self.keys 中的值作为连接参数
                self.keys["gProj"],  # 实例连接名
                "pg8000",
                user=self.keys["gUser"],  # 数据库用户
                password=self.keys["gPass"],  # 数据库密码
                db=self.keys["gDB"],  # 数据库名
                ip_type=ip_type,
            )
            return conn

        # Cloud SQL Python Connector 可以与SQLAlchemy一起使用，
        # 通过 'creator' 参数传递 'getconn' 函数
        pool = sqlalchemy.create_engine(
            "postgresql+pg8000://",
            creator=getconn,
            # 其他连接池参数...
        )
        return pool

# 示例用法 (假设你有一个配置字典)
# if __name__ == "__main__":
#     # 模拟配置信息，实际应用中可能来自配置文件、命令行参数或更安全的存储
#     config = {
#         "gProj": "my-gcp-project:us-central1:my-cloudsql-instance",
#         "gUser": "my_db_user",
#         "gPass": "my_db_password",
#         "gDB": "my_database",
#         "gPrivIP": "USE_PRIVATE_IP_FOR_CLOUD_SQL" # 这是一个环境变量的名称
#     }
#     # 假设环境变量 USE_PRIVATE_IP_FOR_CLOUD_SQL 已经设置或未设置
#     # os.environ["USE_PRIVATE_IP_FOR_CLOUD_SQL"] = "true" # 如果要测试私有IP
#
#     db_connector = DatabaseConnector(config)
#     try:
#         engine = db_connector.connect_with_connector()
#         with engine.connect() as conn:
#             result = conn.execute(sqlalchemy.text("SELECT 1")).scalar()
#             print(f"Connection successful, result: {result}")
#     except KeyError as e:
#         print(f"KeyError occurred: {e}. Please check your configuration and environment variables.")
#     except Exception as e:
#         print(f"An unexpected error occurred: {e}")