如何用SQLAlchemy跨SQL Server迁移数据并维持字符集与排序规则？

问题概况 #

因权限限制需通过Python脚本迁移含阿拉伯语字段的远程SQL Server库到本地：

• 远程库字段排序规则为SQL_Latin1_General_CP1_CI_AS
• Python读取阿拉伯语内容正常，但写入本地库后显示为问号
• 本地库手动修改排序规则为Arabic_CI_AI_KS_WS后可正常插入，但脚本迁移时表排序规则会被自动重置，调整引擎连接参数无效

原问题代码 #

import codecs
import sys
import urllib
import time
import sqlalchemy as sal
import pandas as pd
import contextlib
from sqlalchemy import create_engine, event
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session
import pyodbc

# SETTING UP THE CONNECTIONS ------------------------------------------------------------------------!

remoteServer = 'IP'
remoteDatabase = 'dbo'
remoteTrusted_Connection = 'yes'

localServer = 'IP'
localDatabase = 'dbo'
localUID = 'user'
localPWD = 'pass'

conArgs = {'init_command':"SET @@collation_connection='Arabic_CI_AI_KS_WS'"}

remoteParams = 'DRIVER={ODBC Driver 17 for SQL Server};SERVER=' + remoteServer + ';DATABASE=' + remoteDatabase + ';Trusted_Connection=yes;'
remoteURLib = urllib.parse.quote_plus(remoteParams)
remoteEngine = sal.create_engine("mssql+pyodbc:///?odbc_connect=%s?charset=utf8" % remoteURLib, connect_args=conArgs)

remoteInspect = sal.inspect(remoteEngine)
remoteConn = remoteEngine.connect()

localParams = 'DRIVER={ODBC Driver 17 for SQL Server};SERVER=' + localServer + ';DATABASE=' + localDatabase + ';UID=' + localUID + ';PWD=' + localPWD + ';'
localURLib = urllib.parse.quote_plus(localParams)
localEngine = sal.create_engine("mssql+pyodbc:///?odbc_connect=%s?charset=utf8" % localURLib, connect_args=conArgs)

localInspect = sal.inspect(localEngine)
localConn = localEngine.connect()

remoteTableNames = remoteInspect.get_table_names()

# HERE IS WHERE THE DATA MOVING HAPPENS ------------------------------------------------------------------------!

totalTime = time.time()

for index, tableName in enumerate(remoteTableNames):
    thisTime = time.time()
    print(f'This is table #{index}: {tableName}')

    tableData = pd.read_sql(tableName, remoteConn)
    tableData.to_sql(tableName, localEngine, if_exists='replace', index=False)

    print(f'Completed {tableName} in {time.time() - thisTime:.1f} seconds')

print()
print(f'Done - it took {time.time() - totalTime:.1f} seconds')

# ----------------------------------------------------------------------------------------------------------------------!

remoteConn.close()
localConn.close()

问题根源 #

使用to_sql的if_exists='replace'参数时，pandas会根据读取的远程表数据自动生成本地表结构，此时会直接继承远程表的SQL_Latin1_General_CP1_CI_AS排序规则，而非使用本地库的设置，导致阿拉伯语字符无法正确存储，显示为问号。

解决方案 #

核心修正点 #

1. 避免使用replace模式，改用append模式插入数据，前提是本地已创建好带Arabic_CI_AI_KS_WS排序规则的表结构
2. 手动生成表DDL并替换排序规则，确保本地表创建时就使用正确的字符排序配置
3. 用SQLAlchemy的连接事件确保本地连接始终使用目标排序规则，替代无效的connect_args参数

修改后的代码 #

import urllib
import time
import sqlalchemy as sal
import pandas as pd
import pyodbc

# 连接配置
remoteServer = 'IP'
remoteDatabase = '你的远程数据库名称'  # 修正：原代码误将schema设为数据库名
remoteTrusted_Connection = 'yes'

localServer = 'IP'
localDatabase = '你的本地数据库名称'
localUID = 'user'
localPWD = 'pass'

# 本地连接初始化：设置排序规则
def set_local_collation(conn, cursor):
    cursor.execute("SET collation_connection = 'Arabic_CI_AI_KS_WS'")

# 远程数据库连接
remoteParams = (
    f"DRIVER={{ODBC Driver 17 for SQL Server}};"
    f"SERVER={remoteServer};"
    f"DATABASE={remoteDatabase};"
    f"Trusted_Connection=yes;"
)
remoteURLib = urllib.parse.quote_plus(remoteParams)
remoteEngine = sal.create_engine(f"mssql+pyodbc:///?odbc_connect={remoteURLib}")

remoteInspect = sal.inspect(remoteEngine)
remoteConn = remoteEngine.connect()

# 本地数据库连接：绑定事件设置排序规则
localParams = (
    f"DRIVER={{ODBC Driver 17 for SQL Server}};"
    f"SERVER={localServer};"
    f"DATABASE={localDatabase};"
    f"UID={localUID};"
    f"PWD={localPWD};"
)
localURLib = urllib.parse.quote_plus(localParams)
localEngine = sal.create_engine(f"mssql+pyodbc:///?odbc_connect={localURLib}")

# 每次连接时自动执行排序规则设置
sal.event.listen(localEngine, 'connect', set_local_collation)

localInspect = sal.inspect(localEngine)
localConn = localEngine.connect()

remoteTableNames = remoteInspect.get_table_names()

# 迁移执行逻辑
total_time = time.time()

for index, table_name in enumerate(remoteTableNames):
    start_time = time.time()
    print(f'处理第{index}张表：{table_name}')

    # 读取远程表数据
    table_data = pd.read_sql(table_name, remoteConn)

    # 步骤1：获取远程表DDL并修改排序规则
    with remoteConn.begin():
        ddl_result = remoteConn.execute(f"SELECT OBJECT_DEFINITION(OBJECT_ID('{table_name}'))")
        original_ddl = ddl_result.scalar()
        # 替换排序规则为目标值
        modified_ddl = original_ddl.replace('SQL_Latin1_General_CP1_CI_AS', 'Arabic_CI_AI_KS_WS')
        # 删除本地已存在的表（如果有）
        localConn.execute(f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name}")
        # 创建带正确排序规则的本地表
        localConn.execute(modified_ddl)

    # 步骤2：插入数据到本地表
    table_data.to_sql(table_name, localEngine, if_exists='append', index=False)

    print(f'{table_name}处理完成，耗时{time.time() - start_time:.1f}秒')

print(f'\n全部迁移完成，总耗时{time.time() - total_time:.1f}秒')

# 关闭连接
remoteConn.close()
localConn.close()

关键说明 #

• 原代码中remoteDatabase和localDatabase设置为dbo是错误的，需改为实际的数据库名称（dbo是默认schema）
• 使用SQLAlchemy的connect事件替代connect_args，确保每次连接都能正确设置排序规则
• 手动修改DDL的方式能精准控制表结构的排序规则，避免pandas自动生成表时继承远程库的配置
• 若本地已预先创建好符合要求的表结构，可跳过DDL修改步骤，直接用append模式插入数据

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