排查 mssql-python 中的查询、数据和操作问题

请使用本文来诊断与 mssql-python 驱动程序相关的查询执行、数据类型、性能、事务和批量复制问题。

查询执行问题

未找到表格或对象

症状：

ProgrammingError: [42S02] (208) Invalid object name 'TableName'.

可能的原因和解决方案：

数据库上下文错误

cursor.execute("SELECT DB_NAME()")
print(cursor.fetchone()[0])

模式未指定

cursor.execute("SELECT * FROM dbo.TableName")

表格不存在

cursor.execute("""
    SELECT TABLE_NAME
    FROM INFORMATION_SCHEMA.TABLES
    WHERE TABLE_NAME = 'TableName'
""")

语法错误

症状：

ProgrammingError: [42000] (102) Incorrect syntax near '...'.

解决方案：

在 SQL Server Management Studio（SSMS）中测试 SQL 语句以验证语法。

使用参数化查询代替字符串插值：

# Don't use string interpolation for query parameters.
cursor.execute(f"SELECT * FROM Production.Product WHERE Name = '{name}'")

# Use parameters.
cursor.execute(
    "SELECT * FROM Production.Product WHERE Name = %(name)s",
    {"name": name},
)

参数误差

症状：

ProgrammingError: [07001] Wrong number of parameters

解决方案：

统计占位符和参数。计数必须匹配。

选择正确的参数样式：

# Qmark style: positional parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = ? AND Name LIKE ?",
    (1, "Adjustable%"),
)
print(cursor.fetchone())

# Pyformat style: named parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = %(id)s AND Name LIKE %(name)s",
    {"id": 1, "name": "Adjustable%"},
)
print(cursor.fetchone())

多结果批次中的错误行计数

如果 rowcount 在 nextset() 之后重复第一个语句的计数，则升级驱动。每成功 nextset() 一次，都会刷新新结果的计数。产生行的结果在获取前从 -1 开始。在取数据前检查 cursor.description，因为数据修改结果包含的是计数而不是行。

需要受影响行结果时，请将 SET NOCOUNT 保持关闭。完整示例请参见 “在批处理中跟踪每个结果操作”。

数据类型问题

日期时间转换错误

症状：

DataError: [22007] Invalid datetime format

Solution:

用Pythondatetime对象代替字符串。

from datetime import datetime

cursor.execute("CREATE TABLE #Events (EventDate DATETIME)")

# This value raises an error because the date is invalid.
try:
    cursor.execute(
        "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
        {"event_date": "2024-13-45"},
    )
except Exception as e:
    print(f"Expected error: {e}")

# Use a Python datetime object.
cursor.execute(
    "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
    {"event_date": datetime(2024, 3, 15)},
)
cursor.execute("SELECT EventDate FROM #Events")
print(cursor.fetchone())

时间戳构造器需要微秒参数

如果Timestamp(year, month, day, hour, minute, second)因缺少TypeError参数而引发microsecond，请升级mssql-python。当前驱动接受六参数的 DB-API 构造子，默认 microsecond 为零。更多信息请参见 DB-API 时间戳构造器。

十进制精度问题

症状：

数字似乎被截断或四舍五入。

Solution:

使用 decimal.Decimal 表示精确的数值：

from decimal import Decimal

cursor.execute("CREATE TABLE #PriceDemo (ListPrice DECIMAL(10, 2))")
cursor.execute(
    "INSERT INTO #PriceDemo (ListPrice) VALUES (%(list_price)s)",
    {"list_price": Decimal("19.99")},
)

Unicode 编码问题

症状：

特殊字符会显得杂乱或导致错误。

解决方案：

在数据库中使用 nvarchar 列来记录Unicode数据。

直接传入字符串。驱动程序负责编码：

cursor.execute("CREATE TABLE #UnicodeDemo (Name NVARCHAR(50))")
cursor.execute(
    "INSERT INTO #UnicodeDemo (Name) VALUES (%(name)s)",
    {"name": "日本語"},
)
cursor.execute("SELECT Name FROM #UnicodeDemo")
print(cursor.fetchone())

性能问题

查询执行缓慢

可能的原因和解决方案：

缺失的索引： 检查SSMS中的查询执行计划。

大型结果集： 使用 fetchmany() 代替 fetchall()：

cursor.arraysize = 1000
while True:
    rows = cursor.fetchmany()
    if not rows:
        break
    process_rows(rows)

连接池已禁用： 启用连接池：

import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

大型结果集的内存问题

症状：

Python进程内存不足。

解决方案：

以流式方式返回结果，而不是将所有行都加载到内存中。

cursor.execute("SELECT * FROM LargeTable")
for row in cursor:
    process_row(row)

使用服务器端分页。

page_size = 1000
offset = 0

while True:
    cursor.execute(
        "SELECT * FROM LargeTable ORDER BY ID "
        "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
        (offset, page_size),
    )
    rows = cursor.fetchall()
    if not rows:
        break
    process_rows(rows)
    offset += page_size

交易问题

自动提交模式下的临时表作用域

你在事务中创建的会话临时表（#tablename）在事务回滚时会消失。当自动提交关闭时，这种行为通常会引起混淆，自动提交是默认状态：

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")

# An explicit rollback or an error removes #TempData.
conn.rollback()

# This statement fails with "Invalid object name '#TempData'".
cursor.execute("SELECT * FROM #TempData")

创建临时表后立即提交，或者使用自动提交模式：

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
conn.commit()

cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")
conn.commit()

需要自动提交模式的 DDL 语句，如 CREATE DATABASE，在未完成的事务中会失败。在运行它们之前设置自动提交：

conn.autocommit = True
cursor.execute("CREATE DATABASE TestDB")
conn.autocommit = False

事务未提交

症状：

关闭连接后，数据更改不会持续存在。

Solution:

假设 autocommit=False，即默认值，记为 commit()：

cursor.execute("CREATE TABLE #Products (Name NVARCHAR(100))")
cursor.execute(
    "INSERT INTO #Products (Name) VALUES (%(name)s)",
    {"name": "Widget"},
)
conn.commit()

或者，可以使用自动提交模式：

conn = mssql_python.connect(connection_string, autocommit=True)

死锁错误

症状：

OperationalError: [40001] (1205) Transaction ... was deadlocked on lock resources with another process

Solution:

重试逻辑处理即时失败，但反复出现的死锁表明设计存在问题。捕捉死锁图，分析语句和锁类型。常见的修复方法包括以下更改：

重新排序操作，使竞争事务按相同顺序获取锁。
缩小交易范围。
添加合适的索引以缩短锁定时间。

关于死锁分析的完整攻略，请参见死锁指南。如果你使用Azure SQL 数据库，请参见“分析并防止死锁”。

批量加载问题

批量复制在软件包升级后失败

如果常规查询有效，但 bulkcopy() 或 bulkcopy_arrow() 导入或加载原生批量复制模块失败，请升级驱动并添加依赖：

python -m pip install --upgrade mssql-python
python -m pip check

驱动程序需要与 mssql-python-rs 匹配的运行时。部分升级或使用 --no-deps 进行安装可能会留下不兼容的软件包。如果你使用私有包索引，请镜像这两个伴随包并保留驱动声明的依赖版本锁定。关于当前的固定版本，请参见依赖要求。

不支持的目的地名称

bulkcopy() 和 bulkcopy_arrow() 最多各接受三个标识符部分。像四部分名称这样的名称在发送任何行之前会在元数据查找期间引发 RuntimeError。直接连接到目标服务器，使用单部分、两部分或三部分的表名称。关于括号转义和名称格式错误，请参见目标表名称。

批量复制期间丢失 Unicode 或小数秒

如果批量复制导致 UTF-8 文本损坏，或在加载兼容的 datetime2 值时截断分数秒，请升级驱动。当前的批量复制运行时处理 UTF-8 文本，并保持源类型和目的类型可用的 datetime2 精度。

对 Unicode 文本，请使用 nvarchar，或在 varchar 上使用 UTF-8 排序规则。 Python datetime 值和驱动程序的 Arrow 获取方法限制在微秒级。对于已存储在 Arrow 纳秒时间戳列中的 100 纳秒值，请在不先转换为 Python 的情况下加载到 datetime。更多信息请参见 Unicode 与时间精度以及 Arrow 时间精度。

批量复制期间的约束违规

症状：

RuntimeError: CHECK constraint ... Conflict occurred in database ...
RuntimeError: Cannot insert duplicate key ... violation of PRIMARY KEY constraint

原因：

你的批处理中的数据违反了表约束，比如主键、唯一键、校验键或外键约束。

Solution:

加载数据前请先验证。对于大型数据集，将数据加载到一个预备表中，然后将其合并到目标表中：

cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(100))")
cursor.bulkcopy("##Staging", rows)

# Check for duplicate rows before the merge.
cursor.execute("""
    SELECT s.ID
    FROM ##Staging AS s
    INNER JOIN dbo.Target AS t
        ON s.ID = t.ID
""")
dupes = cursor.fetchall()
if dupes:
    print(f"Skipping {len(dupes)} duplicate rows")

# Insert rows that don't exist in the target.
cursor.execute("""
    INSERT INTO dbo.Target (ID, Name)
    SELECT s.ID, s.Name
    FROM ##Staging AS s
    WHERE NOT EXISTS (
        SELECT 1
        FROM dbo.Target AS t
        WHERE t.ID = s.ID
    )
""")
conn.commit()

关于带有暂存表的更新插入模式，请参阅数据加载和移动模式。

列映射错误

症状：

RuntimeError: Bulk copy failure - column count mismatch

原因：

你数据中的列数和目标表的列数不匹配，或者列的顺序错误。

Solution:

确保你的数据在顺序和计数上与表模式一致：

from decimal import Decimal

cursor.execute("""
    SELECT COLUMN_NAME, DATA_TYPE
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_NAME = 'MyTable'
    ORDER BY ORDINAL_POSITION
""")
for col in cursor.fetchall():
    print(col)

rows = [
    (1, "Widget", Decimal("19.99")),
    (2, "Gadget", Decimal("29.99")),
]
cursor.bulkcopy("dbo.MyTable", rows)

批量复制时出现类型不匹配

症状：

数据加载，但数值被截断、四舍五入或错误。

原因：

Python 的值并不能干净地映射到目标列类型。常见例子包括 float 加载到十进制列中的值，这可能会丢失精度，以及加载到固定长度列的超大字符串。

Solution:

使用与你模式匹配的 Python 类型：

from decimal import Decimal

rows = [
    # Use Decimal for decimal and numeric columns.
    (1, "Widget", Decimal("19.99")),
    # Avoid float values because they can lose precision.
    # (1, "Widget", 19.99),
]
cursor.bulkcopy("dbo.Products", rows)

NumPy 类型绑定失败

症状：

当你使用 NumPy 整数或浮点类型时，参数会无声地失败或引发数据类型错误。

原因：

像 numpy.int32 和 numpy.int64 这样的 NumPy 类型在 NumPy 2.x 中不能通过 isinstance(x, int)。驱动程序的类型推断无法识别它们，导致意外行为。

Solution:

绑定 NumPy 值前先将 NumPy 值转换为原生 Python 类型：

import numpy as np

cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductID = %(product_id)s",
    {"product_id": int(np.int64(42))},
)

for _, row in df.iterrows():
    cursor.execute(
        "INSERT INTO #Orders (ProductID, Qty) "
        "VALUES (%(product_id)s, %(qty)s)",
        {
            "product_id": int(row["ProductID"]),
            "qty": int(row["Qty"]),
        },
    )

对于较大的数据集，可以使用 Arrow 或 Pandas 的集成路径。这些路径内部处理类型转换。

使用临时表进行批量复制

症状：

cursor.bulkcopy("#TempTable", data) 提出 RuntimeError: Invalid object name '#TempTable'。

原因：

bulkcopy() 由于元数据查找的限制，无法解析会话临时表（#tablename）。全局温度表（##tablename）和永久表都有效。

Solution:

可以使用全局临时表或常规暂存表：

# A global temp table is visible to all sessions and is dropped
# when the last session disconnects.
cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("##Staging", rows)

# Alternatively, use a permanent staging table.
cursor.execute("CREATE TABLE dbo.Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("dbo.Staging", rows)

对于你希望使用会话临时表的小型数据集，可以使用 executemany()：

cursor.execute("CREATE TABLE #Staging (ID INT, Name NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Staging (ID, Name) VALUES (?, ?)",
    rows,
)

反馈

此页面是否有帮助？

上次更新时间为 2026-11-05

排查 mssql-python 中的查询、数据和操作问题

查询执行问题

未找到表格或对象

语法错误

参数误差

多结果批次中的错误行计数

数据类型问题

日期时间转换错误

时间戳构造器需要微秒参数

十进制精度问题

Unicode 编码问题

性能问题

查询执行缓慢

大型结果集的内存问题

交易问题

自动提交模式下的临时表作用域

事务未提交

死锁错误

批量加载问题

批量复制在软件包升级后失败

不支持的目的地名称

批量复制期间丢失 Unicode 或小数秒

批量复制期间的约束违规

列映射错误

批量复制时出现类型不匹配

NumPy 类型绑定失败

使用临时表进行批量复制

相关内容

反馈

其他资源