用 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 '...'.

解決方案:

  1. 請在 SQL Server Management Studio(SSMS)中測試 SQL 陳述句以驗證語法。

  2. 使用參數化查詢代替字串插值:

    # 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

解決方案:

  1. 計算佔位符和參數。 數量必須相符。

  2. 選擇正確的參數樣式:

    # 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 編碼問題

症狀:

特殊字元會顯示為亂碼或導致錯誤。

解決方案:

  1. 在你的資料庫中使用 nvarchar 欄位來管理 Unicode 資料。

  2. 直接傳遞字串。 驅動程式負責編碼:

    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 程序會用盡記憶體。

解決方案:

  1. 串流結果,而不是把所有資料列載入記憶體。

    cursor.execute("SELECT * FROM LargeTable")
    for row in cursor:
        process_row(row)
    
  2. 使用伺服器端分頁。

    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"]),
        },
    )

對於較大的資料集,請使用 ArrowPandas 的整合路徑。 這些路徑內部處理型別轉換。

使用暫存資料表進行大量複製

症狀:

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