Python3中从MSSQL导入数据到SQLite时的类型错误及解决方法

你的问题主要有两个核心原因：executemany的用法错误，以及pyodbc返回的Decimal类型与SQLite的适配问题，下面一步步帮你解决：

1. 修复executemany的误用 #

executemany是用来批量插入多行数据的，它需要接收一个包含多个参数组的可迭代对象（比如[(val1, val2), (val3, val4)]）。但你在循环里传入的是单个row（单条记录的元组），这就会触发参数类型错误。

把循环里的c.executemany改成c.execute即可：

for row in rows:
    print('Row Data:')
    print(row)
    print('')
    try:
        # 用execute处理单条记录
        c.execute("INSERT INTO addresses VALUES (?,?,?,?)", row)
        # 获取计数的正确方式
        c.execute("SELECT COUNT(*) FROM addresses")
        rowcount = c.fetchone()[0]
        print(f"当前表中记录数: {rowcount}")
    except sqlite3.Error as er:
        # sqlite3的错误没有Message属性，直接打印错误本身
        print('错误: ', er)

2. 处理Decimal类型适配问题 #

虽然SQLite的NUMERIC类型理论上支持Decimal，但有时候sqlite3驱动可能需要显式适配。你可以通过两种方式解决：

方式一：注册Decimal适配器 #

在连接SQLite后，添加一行代码让sqlite3支持直接插入Decimal类型：

import decimal
tempConn = sqlite3.connect('example.db')
# 注册Decimal适配器，将其转换为SQLite能识别的数值类型
# 若需保留高精度，建议转字符串（SQLite会自动识别为NUMERIC）
sqlite3.register_adapter(decimal.Decimal, str)
# 若对精度要求不高，也可以转float
# sqlite3.register_adapter(decimal.Decimal, lambda d: float(d))

方式二：手动转换Decimal为float/字符串 #

在插入前手动把Decimal类型转换成float或者字符串：

for row in rows:
    # 转换第一个元素（Decimal）为字符串以保留精度
    converted_row = (str(row[0]), row[1], row[2], row[3])
    c.execute("INSERT INTO addresses VALUES (?,?,?,?)", converted_row)

注意：如果你的金额需要高精度，优先选择字符串转换，避免float的精度丢失问题。

3. 补充事务提交 #

别忘了在所有插入操作完成后提交事务，否则数据不会真正写入SQLite数据库：

# 循环结束后提交
tempConn.commit()
# 最后关闭连接
tempConn.close()
conn.close()

完整修正后的代码 #

import sqlite3
import pyodbc
import decimal

# 处理SQLite连接与表创建
tempConn = sqlite3.connect('example.db')
# 注册Decimal适配器，保证高精度
sqlite3.register_adapter(decimal.Decimal, str)
c = tempConn.cursor()
c.execute(
    ''' CREATE TABLE IF NOT EXISTS addresses (
        amt NUMERIC,
        city text,
        statecode text,
        zipcode text
    ) '''
)

# 处理MSSQL连接
conn = pyodbc.connect(
    r'DRIVER={ODBC Driver 11 for SQL Server};'
    r'SERVER=<server>;'
    r'DATABASE=<database>;'
    r'UID=user;'
    r'PWD=password'
)
conn.setdecoding(pyodbc.SQL_CHAR, encoding='utf-8')
conn.setdecoding(pyodbc.SQL_WCHAR, encoding='utf-8')
conn.setencoding(encoding='utf-8')
cursor = conn.cursor()
cursor.execute("SELECT TOP 10 fd.Amt, fd.City, fd.StateCode, fd.ZipCode FROM <large table> AS fd")
rows = cursor.fetchall()

# 插入数据到SQLite
for row in rows:
    print('Row Data:')
    print(row)
    print('')
    try:
        c.execute("INSERT INTO addresses VALUES (?,?,?,?)", row)
        c.execute("SELECT COUNT(*) FROM addresses")
        rowcount = c.fetchone()[0]
        print(f"当前记录数: {rowcount}")
    except sqlite3.Error as er:
        print('错误信息: ', er)

# 提交事务并关闭连接
tempConn.commit()
tempConn.close()
conn.close()

这样应该就能解决你遇到的参数类型错误，顺利完成数据导入了。

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