使用本文來診斷 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:
用 Python datetime 物件代替字串。
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 Database,請參閱「分析與防止死結」。
散裝載重問題
批量複製期間的限制違規
症狀:
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 的值不會乾淨俐落地對應到目標欄位類型。 常見的例子包括載入到 decimal 欄位中的 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 類型numpy.int64在 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,
)