使用SQLAlchemy和Pandas高效管理多数据库连接与数据迁移

import pandas as pd
from sqlalchemy import create_engine, text
import pymysql # 确保已安装pymysql作为MySQL的DBAPI驱动

# --- 远程数据库配置 ---
remote_hostname = "remote.server.com" # 替换为您的远程主机名
remote_username = "remote_user"      # 替换为您的远程用户名
remote_password = "remote_pass"      # 替换为您的远程密码
remote_database = "remote_db"        # 替换为您的远程数据库名

# 创建远程数据库引擎
remote_engine = create_engine(
    f"mysql+pymysql://{remote_username}:{remote_password}@{remote_hostname}/{remote_database}"
)

# --- 本地数据库配置 ---
local_hostname = "localhost"         # 替换为您的本地主机名
local_username = "local_user"        # 替换为您的本地用户名
local_password = "local_pass"        # 替换为您的本地密码
local_database = "local_db"          # 替换为您的本地数据库名

# 创建本地数据库引擎
local_engine = create_engine(
    f"mysql+pymysql://{local_username}:{local_password}@{local_hostname}/{local_database}"
)

print("数据库引擎已成功创建。")

# 定义要从远程数据库读取的SQL查询
getcommand = text("SELECT * FROM your_remote_table") # 替换为您的远程表名

df = None # 初始化DataFrame

# 使用with语句安全地获取远程数据库连接
with remote_engine.connect() as remote_conn:
    print("已连接到远程数据库，开始读取数据...")
    df = pd.read_sql(getcommand, remote_conn)
    print(f"从远程数据库读取了 {len(df)} 条数据。")

# 此时，remote_conn 已被自动关闭或返回到连接池
if df is not None:
    print("读取到的数据预览：")
    print(df.head())
else:
    print("未能从远程数据库读取到数据。")

# 示例：对DataFrame进行一些简单的处理
# df['new_column'] = df['existing_column'] * 2
# df = df.dropna() # 移除缺失值
print("数据处理步骤完成（如果适用）。")

if df is not None and not df.empty:
    # 使用with语句安全地获取本地数据库连接并开启事务
    with local_engine.connect() as local_conn, local_conn.begin():
        print("已连接到本地数据库，开始写入数据...")
        df.to_sql(
            name="your_local_table", # 替换为您的本地目标表名
            con=local_conn,
            if_exists="replace",     # 如果表存在则替换，可选 'append', 'fail'
            index=False              # 不将DataFrame的索引写入数据库
        )
        print(f"成功将 {len(df)} 条数据写入本地数据库。")
    # 此时，local_conn 已被自动关闭或返回到连接池，事务已提交或回滚。
else:
    print("没有数据可写入本地数据库。")

print("数据迁移过程完成。")

import pandas as pd
from sqlalchemy import create_engine, text
import pymysql # 确保已安装pymysql

# --- 远程数据库配置 ---
remote_hostname = "remote.server.com"
remote_username = "remote_user"
remote_password = "remote_pass"
remote_database = "remote_db"

# 创建远程数据库引擎
remote_engine = create_engine(
    f"mysql+pymysql://{remote_username}:{remote_password}@{remote_hostname}/{remote_database}"
)

# --- 本地数据库配置 ---
local_hostname = "localhost"
local_username = "local_user"
local_password = "local_pass"
local_database = "local_db"

# 创建本地数据库引擎
local_engine = create_engine(
    f"mysql+pymysql://{local_username}:{local_password}@{local_hostname}/{local_database}"
)

# 定义要从远程数据库读取的SQL查询
getcommand = text("SELECT * FROM your_remote_table") # 替换为您的远程表名

df = None # 初始化DataFrame

try:
    # 1. 从远程数据库读取数据
    print("正在连接远程数据库并读取数据...")
    with remote_engine.connect() as remote_conn:
        df = pd.read_sql(getcommand, remote_conn)
    print(f"成功从远程数据库读取 {len(df)} 条数据。")
    print("数据预览：\n", df.head())

    # 2. 数据处理（示例，根据需要修改）
    # df['processed_column'] = df['original_column'] * 10
    # df = df.dropna()
    print("数据处理完成（如果适用）。")

    # 3. 将处理后的数据写入本地数据库
    if not df.empty:
        print("正在连接本地数据库并写入数据...")
        with local_engine.connect() as local_conn, local_conn.begin():
            df.to_sql(
                name="your_local_table", # 替换为您的本地目标表名
                con=local_conn,
                if_exists="replace",
                index=False
            )
        print(f"成功将 {len(df)} 条数据写入本地数据库表 'your_local_table'。")
    else:
        print("DataFrame为空，没有数据写入本地数据库。")

except Exception as e:
    print(f"发生错误: {e}")

finally:
    print("数据迁移过程结束。")
    # 显式关闭引擎（虽然通常在脚本结束时会自动清理，但明确写出有助于理解）
    # remote_engine.dispose()
    # local_engine.dispose()
    # print("数据库引擎已释放。")