排查 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())

数据类型问题

日期时间转换错误

症状：

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())

十进制精度问题

症状：

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

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 数据库，请参见“分析并防止死锁”。

批量加载问题

批量复制期间的约束违规

症状：

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,
)

反馈

此页面是否有帮助？

Last updated on 2026-09-21

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

查询执行问题

未找到表格或对象

语法错误

参数误差

数据类型问题

日期时间转换错误

十进制精度问题

Unicode 编码问题

性能问题

查询执行缓慢

大型结果集的内存问题

交易问题

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

事务未提交

死锁错误

批量加载问题

批量复制期间的约束违规

列映射错误

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

NumPy 类型绑定失败

使用临时表进行批量复制

相关内容

反馈

其他资源