如何用SQLAlchemy+Pandas将循环元组转为元组的元组并生成DataFrame

我正在用SQLAlchemy执行Snowflake表的SQL查询，当前代码如下：

for x in list:
    results = cursor.execute(f"SELECT TABLE_NAME as TABLE_NAME, 'TABLE_SCHEMA as TABLE_SCHEMA, MAX(DATE) as DATE  FROM {database}.{schema}.{table}")
    for result in results:
        print(result)

循环每次迭代返回一个元组，示例：

('SCHEMA_1','TABLE_1','DATE_1')
('SCHEMA_2','TABLE_2','DATE_2')
('SCHEMA_3','TABLE_3','DATE_3')

需要把这些元组组成元组的元组，转成DataFrame后写回Snowflake。要求：不能用snowflake-connector-python[pandas]，只能用SQLAlchemy + Pandas实现。

步骤1：修正语法错误并收集结果 #

原SQL语句中'TABLE_SCHEMA多了一个单引号，会引发语法错误，先修正该问题。然后初始化列表收集所有迭代返回的元组，最后转换为元组的元组。

步骤2：转换为Pandas DataFrame #

使用Pandas的DataFrame构造函数，传入收集到的元组集合，同时指定与查询一致的列名。

步骤3：通过SQLAlchemy写入Snowflake #

利用Pandas的to_sql方法，结合SQLAlchemy引擎连接，将DataFrame写入目标Snowflake表。

完整代码示例 #

import pandas as pd
from sqlalchemy import create_engine

# 假设已通过SQLAlchemy配置好Snowflake连接引擎 engine
all_results = []

# 遍历处理并修正SQL语法错误
for x in list:
    query = f"SELECT TABLE_NAME as TABLE_NAME, TABLE_SCHEMA as TABLE_SCHEMA, MAX(DATE) as DATE FROM {database}.{schema}.{table}"
    results = cursor.execute(query)
    # 将当前迭代的所有元组加入结果列表
    all_results.extend(results)

# 转换为元组的元组
tuple_of_tuples = tuple(all_results)

# 生成DataFrame
df = pd.DataFrame(tuple_of_tuples, columns=["TABLE_NAME", "TABLE_SCHEMA", "DATE"])

# 写入Snowflake目标表
df.to_sql(
    name="target_table",
    con=engine,
    schema="target_schema",
    database="target_database",
    if_exists="append",  # 可根据需求替换为replace/fail
    index=False
)

关键说明 #

• 用extend而非append收集结果，避免产生嵌套列表，确保每个元组单独存入列表
• to_sql依赖已配置好的SQLAlchemy Snowflake引擎，需提前正确设置账户、仓库、角色等连接参数
• if_exists参数需根据业务需求选择，控制表已存在时的写入策略

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