解决psycopg2连接AWS Redshift Serverless超时问题

import psycopg2
import os
from dotenv import load_dotenv

# 推荐：从环境变量加载敏感信息，而不是硬编码
# 为此，请确保您的环境中已设置 .env 文件或直接设置环境变量
# 例如：
# export REDSHIFT_ENDPOINT="default-workgroup.xxxx.ap-south-1.redshift-serverless.amazonaws.com"
# export REDSHIFT_PORT="5439"
# export REDSHIFT_DATABASE="dev"
# export REDSHIFT_USER="admin"
# export REDSHIFT_PASSWORD="your_redshift_password"
load_dotenv() # 加载 .env 文件中的环境变量

class RedshiftConnector:
    def __init__(self):
        # 从环境变量获取连接参数
        self.redshift_endpoint = os.getenv('REDSHIFT_ENDPOINT')
        self.redshift_port = os.getenv('REDSHIFT_PORT', '5439') # 默认5439
        self.redshift_database = os.getenv('REDSHIFT_DATABASE')
        self.redshift_user = os.getenv('REDSHIFT_USER')
        self.redshift_password = os.getenv('REDSHIFT_PASSWORD')

        # 检查所有必要的参数是否都已提供
        if not all([self.redshift_endpoint, self.redshift_database, 
                    self.redshift_user, self.redshift_password]):
            raise ValueError("Redshift连接参数未完全通过环境变量提供。请检查REDSHIFT_ENDPOINT, REDSHIFT_DATABASE, REDSHIFT_USER, REDSHIFT_PASSWORD。")

    def connect(self):
        """
        建立与Redshift Serverless的连接。
        """
        print(f"尝试连接到Redshift Serverless: {self.redshift_endpoint}:{self.redshift_port}/{self.redshift_database}")
        try:
            redshift_conn = psycopg2.connect(
                host=self.redshift_endpoint,
                port=self.redshift_port,
                database=self.redshift_database,
                user=self.redshift_user,
                password=self.redshift_password
            )
            print("成功连接到Redshift Serverless!")
            return redshift_conn
        except psycopg2.OperationalError as e:
            print(f"连接Redshift Serverless失败: {e}")
            print("请检查网络连接、AWS安全组配置（确保开放端口5439）以及数据库凭证。")
            raise # 重新抛出异常，以便调用者处理

    def execute_query(self, query, conn, fetch_results=False):
        """
        执行SQL查询。
        :param query: 要执行的SQL语句。
        :param conn: psycopg2数据库连接对象。
        :param fetch_results: 如果为True，则尝试获取查询结果。
        :return: 如果fetch_results为True，则返回查询结果；否则返回None。
        """
        try:
            with conn.cursor() as cursor:
                cursor.execute(query)
                if fetch_results and query.strip().upper().startswith('SELECT'):
                    results = cursor.fetchall()
                    print("查询执行成功并获取到结果。")
                    return results
                else:
                    conn.commit() # 对于DML/DDL操作，需要提交事务
                    print("查询执行成功。")
                    return None
        except Exception as e:
            print(f"执行查询失败: {e}")
            conn.rollback() # 发生错误时回滚事务
            raise

# 示例使用
if __name__ == "__main__":
    connector = RedshiftConnector()
    connection = None
    try:
        connection = connector.connect()

        # 示例：创建表（如果不存在）
        create_table_query = """
        CREATE TABLE IF NOT EXISTS tutorial_data (
            id INT IDENTITY(1,1),
            item_name VARCHAR(255),
            item_value INT,
            created_at TIMESTAMP DEFAULT GETDATE()
        );
        """
        connector.execute_query(create_table_query, connection)

        # 示例：插入数据
        insert_data_query = """
        INSERT INTO tutorial_data (item_name, item_value) VALUES ('Sample Item A', 101);
        INSERT INTO tutorial_data (item_name, item_value) VALUES ('Sample Item B', 202);
        """
        connector.execute_query(insert_data_query, connection)

        # 示例：查询数据
        select_data_query = "SELECT * FROM tutorial_data ORDER BY id DESC LIMIT 5;"
        results = connector.execute_query(select_data_query, connection, fetch_results=True)
        if results:
            print("\n查询结果:")
            for row in results:
                print(row)

    except ValueError as ve:
        print(f"配置错误: {ve}")
    except psycopg2.OperationalError:
        print("请根据上述教程检查您的网络和安全组配置。")
    except Exception as e:
        print(f"发生未预期错误: {e}")
    finally:
        if connection:
            connection.close()
            print("数据库连接已关闭。")