基于Pandas字典与条件分支实现多SQL查询按需执行

首先，我们需要根据my_schema和my_table的取值过滤出需要执行的模式（schema）和表（table），然后为每个有效的组合生成对应的SQL语句并执行。以下是符合需求的完整代码：

import datetime

# 定义模式与表的映射字典
my_dict = {
    'ETL_A': ['TABLE_001', 'TABLE_002', 'TABLE_003'],
    'TEST1': ['TABLE_TEST', 'TEST001', 'SAMPLE_TEST'],
    'TEST2': ['TABLE_TEST2', 'TEST2', 'SAMPLE_TEST2'],
    'MY_PRIVATE_SHEMA': ['WWW_001', 'LTT_222']
}

# 配置参数（可根据需求修改）
my_schema = 'ETL_A'  # 可设为None
my_table = 'TABLE_001'  # 可设为None

# 初始化数据库游标（假设已建立数据库连接并获取游标）
# cur = conn.cursor()

# 获取初始时间戳（非全量执行时使用统一时间戳）
timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')

# 确定需要处理的模式列表
schemas_to_process = []
if my_schema is None:
    schemas_to_process = list(my_dict.keys())
else:
    # 仅处理存在的模式
    if my_schema in my_dict:
        schemas_to_process = [my_schema]
    else:
        print(f"警告：模式 {my_schema} 不存在于字典中，跳过处理")

# 遍历每个模式并处理对应表
for schema in schemas_to_process:
    tables_in_schema = my_dict[schema]
    
    # 确定当前模式下需要处理的表列表
    tables_to_process = []
    if my_table is None:
        tables_to_process = tables_in_schema.copy()
    else:
        # 仅处理存在于当前模式的表
        if my_table in tables_in_schema:
            tables_to_process = [my_table]
        else:
            print(f"警告：表 {my_table} 不存在于模式 {schema} 中，跳过处理")
    
    # 生成并执行SQL语句
    for table in tables_to_process:
        # 全量执行时，为每个查询生成最新时间戳
        if my_schema is None:
            timestamp = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        # 构建SQL语句（注意：原语句会覆盖原表，建议添加后缀避免数据丢失）
        query = f"CREATE TABLE {schema}.{table} SELECT * FROM {schema}.{table} at(timestamp => '{timestamp}'::timestamp);"
        # 若需创建备份表可改为：
        # query = f"CREATE TABLE {schema}.{table}_backup_{timestamp.replace(' ', '_').replace(':', '')} SELECT * FROM {schema}.{table} at(timestamp => '{timestamp}'::timestamp);"
        
        print(f"执行SQL：{query}")
        cur.execute(query)

# 提交事务（根据数据库连接要求）
# conn.commit()

关键逻辑说明 #

1. 模式过滤：

   • 当my_schema为None时，处理字典中所有模式
   • 当my_schema指定时，仅处理该模式（若存在于字典中）

2. 表过滤：

   • 当my_table为None时，处理当前模式下所有表
   • 当my_table指定时，仅处理该表（若存在于当前模式中）

3. 时间戳处理：

   • 全量执行（my_schema为None）时，为每个查询生成最新时间戳
   • 指定模式执行时，使用统一初始时间戳

4. 异常处理：

   • 对不存在的模式或表添加了警告提示，避免无意义的执行错误

注意事项 #

• 原SQL语句会直接创建与原表同名的表，可能导致数据覆盖，建议添加备份后缀（如示例中注释的写法）
• 需确保数据库游标cur已正确初始化并建立连接
• 执行完成后需根据数据库要求提交事务（如conn.commit()）

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