使用SQLAlchemy从SQL Server迁移数据到MariaDB时遇类型错误

问题背景 #

尝试用SQLAlchemy将SQL Server的customers表数据迁移到MariaDB，两张表结构基本一致：

MariaDB表结构 #

CREATE TABLE IF NOT EXISTS customers
(
    id           INT PRIMARY KEY,
    first_name   VARCHAR(50)  NOT NULL,
    last_name    VARCHAR(50)  NOT NULL,
    email        VARCHAR(50)  NOT NULL UNIQUE,
    phone_number VARCHAR(15)  NOT NULL,
    address      VARCHAR(150) NOT NULL,
    city         VARCHAR(50)  NOT NULL,
    state        VARCHAR(50)  NOT NULL,
    zip_code     VARCHAR(10)  NOT NULL,
    create_date  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    update_date  TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

SQL Server表结构 #

CREATE TABLE customers
(
    id           INT PRIMARY KEY IDENTITY (1,1),
    first_name   VARCHAR(50)  NOT NULL,
    last_name    VARCHAR(50)  NOT NULL,
    email        VARCHAR(50)  NOT NULL UNIQUE,
    phone_number VARCHAR(15)  NOT NULL,
    address      VARCHAR(150) NOT NULL,
    city         VARCHAR(50)  NOT NULL,
    state        VARCHAR(50)  NOT NULL,
    zip_code     VARCHAR(10)  NOT NULL,
    create_date  DATETIME DEFAULT CURRENT_TIMESTAMP,
    update_date  DATETIME DEFAULT CURRENT_TIMESTAMP
);

迁移代码 #

def main():
    # Connection strings
    sql_server_conn_str = CONFIG['connectionStrings']['sqlServer']
    maria_conn_str = CONFIG['connectionStrings']['mariaDb']

    # create SQLAlchemy engine for SQL Server and MariaDB
    sql_server_engine = create_engine(sql_server_conn_str)
    maria_engine = create_engine(maria_conn_str)

    # create connection for both db
    sql_server_conn = sql_server_engine.connect()
    maria_conn = maria_engine.connect();

    # create SQLAlchemy MetaData objects for SQL Server and MariaDB
    sql_server_metadata = MetaData()
    maria_metadata = MetaData()

    # reflect the SQL Server database schema into the MetaData object
    sql_server_metadata.reflect(bind=sql_server_engine)

    # create Table objects for each SQL Server table
    customers_sql_server = Table('customers', sql_server_metadata, autoload=True, autoload_with=sql_server_engine)

    # reflect the MariaDB database schema into the MetaData object
    maria_metadata.reflect(bind=maria_engine)

    # create Table objects for each MariaDB table
    customers_maria = Table('customers', maria_metadata, autoload=True, autoload_with=maria_engine)

    # select all rows from the customers table in SQL Server
    select_customers_sql_server = select(customers_sql_server)

    # execute the select query and fetch all rows
    result_proxy = sql_server_conn.execute(select_customers_sql_server)
    customers_data = result_proxy.fetchall()

    # insert the rows into the customers table in MariaDB
    tuples_to_insert = [tuple(row) for row in customers_data]
    maria_conn.execute(customers_maria.insert(), tuples_to_insert)

报错信息 #

Traceback (most recent call last):   ...\main.py", line 48, in <module>
        main()   File "...\main.py", line 35, in main
        maria_conn.execute(customers_maria.insert(), tuples_to_insert)   File "...\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 1413, in execute
        return meth(
               ^^^^^   File "...\venv\Lib\site-packages\sqlalchemy\sql\elements.py", line 483, in
    _execute_on_connection
        return connection._execute_clauseelement(
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^   File "...\venv\Lib\site-packages\sqlalchemy\engine\base.py", line 1613, in
    _execute_clauseelement
        keys = sorted(distilled_parameters[0])
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ 

TypeError: '<' not supported between instances of 'str' and 'int'

错误原因 #

报错是因为SQLAlchemy处理批量插入时，同时收到了两种参数格式：从SQL Server取出的元组是位置参数，但insert()方法默认期望批量插入的是字典列表（关键字参数）。内部处理时，参数键同时存在字符串（列名）和数字（索引）类型，排序时触发类型不兼容的错误。

解决方法 #

把从SQL Server取出的行数据转换成字典（键对应列名），让SQLAlchemy能准确匹配MariaDB表的字段，避免参数类型混乱。

修改后的代码 #

def main():
    # Connection strings
    sql_server_conn_str = CONFIG['connectionStrings']['sqlServer']
    maria_conn_str = CONFIG['connectionStrings']['mariaDb']

    # create SQLAlchemy engine for SQL Server and MariaDB
    sql_server_engine = create_engine(sql_server_conn_str)
    maria_engine = create_engine(maria_conn_str)

    # create connection for both db
    sql_server_conn = sql_server_engine.connect()
    maria_conn = maria_engine.connect()

    # create SQLAlchemy MetaData objects for SQL Server and MariaDB
    sql_server_metadata = MetaData()
    maria_metadata = MetaData()

    # reflect the SQL Server database schema into the MetaData object
    sql_server_metadata.reflect(bind=sql_server_engine)

    # create Table objects for each SQL Server table
    customers_sql_server = Table('customers', sql_server_metadata, autoload=True, autoload_with=sql_server_engine)

    # reflect the MariaDB database schema into the MetaData object
    maria_metadata.reflect(bind=maria_engine)

    # create Table objects for each MariaDB table
    customers_maria = Table('customers', maria_metadata, autoload=True, autoload_with=maria_engine)

    # select all rows from the customers table in SQL Server
    select_customers_sql_server = select(customers_sql_server)

    # execute the select query and fetch all rows, convert to dictionaries
    result_proxy = sql_server_conn.execute(select_customers_sql_server)
    # 将行数据转为字典，键为列名
    customers_data = [dict(row) for row in result_proxy]

    # insert the rows into the customers table in MariaDB
    maria_conn.execute(customers_maria.insert(), customers_data)
    # 提交事务，确保数据持久化
    maria_conn.commit()

补充说明 #

1. 替换tuple(row)为dict(row)：字典格式能明确匹配字段，即使后续表结构调整（如列顺序变化）也不会出错，比元组更健壮。
2. 添加maria_conn.commit()：部分SQLAlchemy引擎默认不自动提交事务，必须手动提交才能完成数据插入。
3. 若确认两张表列顺序完全一致，也可使用maria_conn.execute(customers_maria.insert().values(tuples_to_insert))插入元组列表，但字典方式更推荐。

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