使用 mssql-python 的自訂型別轉換器

預設情況下，mssql-python 驅動程式會將 Microsoft SQL 資料型態轉換為適當的 Python 型別（參見資料型態映射）。 輸出轉換系統允許你針對特定 SQL 類型覆蓋此行為，啟用：

• 自訂型別轉換（例如，將 money 轉換為 float，而非 Decimal）
• 與第三方函式庫整合（例如，將空間資料轉換為 Shapely 物件）
• 企業專用格式（例如，日期轉換為自訂字串格式）

註冊輸出轉換器

用 add_output_converter() 來註冊一個轉換函數。 索引鍵可以是與 type_code 中的 cursor.description 相符的 Python 類型，或是整數 ODBC SQL 類型代碼，例如 mssql_python.SQL_DECIMAL：

import mssql_python
from datetime import datetime

def my_datetime_converter(value):
    """Convert datetime to a custom display string."""
    if value is None:
        return None
    return value.strftime("%B %d, %Y at %I:%M %p")

conn = mssql_python.connect(connection_string)
conn.add_output_converter(datetime, my_datetime_converter)

cursor = conn.cursor()
cursor.execute("SELECT CAST('2026-07-16T10:30:00' AS DATETIME2)")
print(cursor.fetchval())  # July 16, 2026 at 10:30 AM

請選擇 Python 類型的金鑰或 SQL 類型的程式碼金鑰

這兩種關鍵形式在選擇欄位的精確度上有所不同：

• Python 型別金鑰適用於所有對應到該 Python 型別的 SQL 型別。 為 Decimal 註冊轉換器會轉換 decimal、numeric、money 和 smallmoney 欄位。

• 以 整數 SQL 型別代碼 為索引鍵，會對應到資料行所回報的 ODBC 型別代碼；後者範圍較窄，但不一定總是單一的 SQL 型別。 SQL_NUMERIC 僅轉換 數值 欄位。 SQL_DECIMAL 會轉換 decimal、money 和 smallmoney 欄位，因為這三者都回報相同的代碼。

當 Python 類型的鍵比你想要的寬，或是從 pyodbc 移植程式碼時，可以使用整數鍵。 當你想要一個轉換器涵蓋一整類 SQL 類型時，可以使用 Python 類型的金鑰。

import mssql_python

def tag_decimal(value):
    return f"D:{value}"

conn.add_output_converter(mssql_python.SQL_DECIMAL, tag_decimal)

cursor.execute("""
    SELECT CAST(12.34 AS decimal(10,2)),
           CAST(56.78 AS numeric(10,2)),
           CAST(90.12 AS money)
""")
print(cursor.fetchone())  # ('D:12.34', Decimal('56.78'), 'D:90.1200')

轉換器解析度階數

對於每一欄，駕駛員最多選擇一個轉換器，順序如下：

1. 該轉換器已註冊該欄位的 ODBC SQL 整數類型代碼。

2. 在 cursor.description 中為 Python 類型註冊的轉換器。

3. 轉換器會註冊 SQL_WVARCHAR，但僅當欄位的 Python 類型為 str 或 bytes時。

轉換器函式簽章

當你註冊一個帶有 Python 類型鍵的轉換器時，你傳遞給add_output_converter()的類型必須符合以下 Python 類型之一。 轉換器接收欄位顯示驅動程式傳遞給函式的內容：

from decimal import Decimal

def decimal_converter(value: Decimal | None) -> float | None:
    if value is None:
        return None
    # value is a Decimal object for numeric/money types
    return float(value)

管理轉換器

利用這些方法檢查或移除已註冊在連接上的轉換器。

取得現有轉換器

取得目前為某個型別註冊的轉換器函式：

converter = conn.get_output_converter(str)
if converter:
    print(f"Converter registered: {converter}")
else:
    print("Using default conversion")

拆除轉換器

取消註冊一個轉換器，使驅動程式回復到該類型的預設轉換：

conn.remove_output_converter(str)
# str columns now use default conversion

清除所有轉換器

重設所有註冊轉換器以使用 Microsoft SQL 預設的類型映射：

conn.clear_output_converters()
# All types now use default conversion

常見的轉換器模式

這些範例展示了最常用的轉換器功能。

將 VARCHAR 轉換為大寫

字串型轉換器接收 UTF-16LE 編碼的原始位元組：

def uppercase_converter(value):
    if value is None:
        return None
    return value.decode('utf-16-le').upper()

conn.add_output_converter(str, uppercase_converter)

cursor = conn.cursor()
cursor.execute("SELECT 'hello world'")
print(cursor.fetchval())  # HELLO WORLD

將貨幣兌換成流通股

預設情況下，驅動程式會回傳小數位/數值型別為 decimal.Decimal。 轉換成浮動：

from decimal import Decimal

def money_to_float(value):
    if value is None:
        return None
    return float(value)

conn.add_output_converter(Decimal, money_to_float)

cursor = conn.cursor()
cursor.execute("SELECT CAST(19.99 AS MONEY)")
result = cursor.fetchval()
print(type(result))  # <class 'float'>

Parse JSON data

Microsoft SQL Server 可以將 JSON 儲存為文字。 自動解析成 Python 物件：

import json

def json_converter(value):
    if value is None:
        return None
    text = value.decode('utf-16-le')
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return text  # Return as string if not valid JSON

conn.add_output_converter(str, json_converter)

cursor = conn.cursor()
cursor.execute("SELECT '{\"name\": \"Widget\", \"price\": 19.99}'")
data = cursor.fetchval()
print(data['name'])  # Widget

自訂日期時間格式

將 datetime 轉換為 ISO 字串格式：

from datetime import datetime

def datetime_to_iso(value):
    if value is None:
        return None
    return value.isoformat()

conn.add_output_converter(datetime, datetime_to_iso)

cursor = conn.cursor()
cursor.execute("SELECT TOP 1 ModifiedDate FROM Production.Product")
print(cursor.fetchval())  # 2014-02-08T10:01:36.827000

安全性考慮

# DANGEROUS - never do this
def unsafe_example(user_converter_code):
    converter_func = eval(user_converter_code)  # Security risk!
    conn.add_output_converter(str, converter_func)

範例：空間資料整合

將空間資料整合至 Shapely 函式庫：

from shapely import wkb

def geometry_converter(value):
    """Convert WKB binary to Shapely geometry object."""
    if value is None:
        return None
    return wkb.loads(value)

conn.add_output_converter(bytes, geometry_converter)

# Query must use STAsBinary() to get standard WKB format
cursor = conn.cursor()
cursor.execute(
    "SELECT SpatialLocation.STAsBinary() FROM Person.Address "
    "WHERE SpatialLocation IS NOT NULL AND City = 'Seattle'"
)
for row in cursor:
    shape = row[0]
    print(f"Point: ({shape.x:.4f}, {shape.y:.4f})")

範例：自訂列格式

建立資料類別轉換器：

from dataclasses import dataclass

@dataclass
class Product:
    id: int
    name: str
    price: float

def fetch_products_as_dataclass(cursor):
    """Convert rows to Product dataclass instances."""
    cursor.execute(
        "SELECT ProductID, Name, ListPrice FROM Production.Product "
        "WHERE ListPrice > 0"
    )
    results = []
    for row in cursor:
        results.append(Product(
            id=row.ProductID,
            name=row.Name,
            price=float(row.ListPrice)
        ))
    return results

products = fetch_products_as_dataclass(cursor)
for p in products[:5]:
    print(f"{p.name}: ${p.price:.2f}")

轉換器的範圍與生命週期

• 以連線為範圍：驅動程式會為每個連線各自註冊轉換器，而不是全域註冊。
• 持續性：轉換器會持續運作，直到你移除它們或關閉連接為止。
• 非繼承：新連接不會繼承其他連接的轉換器。

conn1 = mssql_python.connect(connection_string)
conn1.add_output_converter(str, uppercase_converter)

conn2 = mssql_python.connect(connection_string)
# conn2 does NOT have the converter registered

效能考量

• 驅動程式會針對已註冊型別的 每個值 呼叫轉換器。
• 保持轉換器功能有效率。
• 對於大量查詢，請考慮是否需要轉換。
• 如果轉換器會影響吞吐量，請剖析效能。

import time

def slow_converter(value):
    time.sleep(1)  # Avoid time consuming routines like this - adds 1 second for each value processed!
    return str(value) if value else None

相關內容

• mssql-python 的資料型別映射
• 使用 mssql-python 擷取資料
• 用 mssql-python 管理連線