如何在Python(pandas)中创建数据库连接循环以对多库执行同一查询

方案1：直接复用你已创建的连接实例 #

你只需要把现有数据库引擎放入列表，遍历执行即可：

import pandas as pd
from sqlalchemy import create_engine

# 你已创建的数据库连接实例列表
db_engine_list = [db1, db2, db3]
# 统一执行的查询语句
query_sql = """select * from table """

# 循环执行查询+写入操作
for current_engine in db_engine_list:
    df = pd.read_sql(query_sql, con=current_engine)
    df.to_sql('table2', con=dwh_engine, if_exists='append', index=False)

方案2：优化连接配置，避免重复代码 #

如果后续需要新增数据库，推荐用配置模板动态生成连接，减少重复代码：

import pandas as pd
from sqlalchemy import create_engine

# 配置项：需要遍历的数据库名、公共连接串模板
db_name_list = ["db1", "db2", "db3"]
conn_template = "mssql+pymssql://login:pass@host:1433/{}"
query_sql = """select * from table """

for db_name in db_name_list:
    # 动态生成当前数据库的连接引擎
    current_engine = create_engine(conn_template.format(db_name))
    # 执行查询
    df = pd.read_sql(query_sql, con=current_engine)
    # 写入目标库
    df.to_sql('table2', con=dwh_engine, if_exists='append', index=False)
    # 关闭当前连接，释放资源
    current_engine.dispose()

可选：添加异常捕获逻辑 #

避免单个数据库连接/查询失败打断整个循环：

for db_name in db_name_list:
    try:
        current_engine = create_engine(conn_template.format(db_name))
        df = pd.read_sql(query_sql, con=current_engine)
        df.to_sql('table2', con=dwh_engine, if_exists='append', index=False)
        current_engine.dispose()
        print(f"数据库{db_name}同步完成")
    except Exception as e:
        print(f"数据库{db_name}同步失败，错误信息：{str(e)}")

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