從 pyodbc 遷移到 mssql-python

mssql-python 驅動程式是 Microsoft 為 Microsoft SQL 提供的第一方 Python 驅動程式。如果你偏好 Microsoft 維護的驅動程式選項，它提供：

沒有外部 ODBC 驅動程式依賴。
內建連線池。
現代 Python 3.10+ 支援。
原生 Microsoft Entra 驗證。

主要差異

Feature	pyodbc	MSSQL-Python
參數樣式	`qmark` （`?`）	`qmark` （`?`）以及 `pyformat` （`%(name)s`）
需要 ODBC 驅動程式	是的	No
連線池化	外部	內建
最低限度的 Python	3.6	3.10
`callproc()`	Supported	未實作
自動提交預設	Off	Off

基本遷移步驟

以下步驟涵蓋將 pyodbc 應用程式遷移至 mssql-python 的主要變更。

1. 更新匯入

將 pyodbc 匯入替換為 mssql_python：

變更前（pyodbc）：

import pyodbc

在（mssql-python）之後：

import mssql_python

2. 更新連接字串

移除關鍵字 DRIVER= 並更新認證方法：

之前（pyodbc，需要 ODBC 驅動）：

conn = pyodbc.connect(
    "DRIVER={ODBC Driver 18 for SQL Server};"
    "SERVER=localhost;"
    "DATABASE=AdventureWorks2022;"
    "Trusted_Connection=yes;"
)

之後（mssql-python，無需驅動程式，使用 Microsoft Entra 認證）：

conn = mssql_python.connect(
    "Server=<server>.database.windows.net;"
    "Database=<database>;"
    "Authentication=ActiveDirectoryDefault;"
    "Encrypt=yes;"
)

3. 保持查詢原樣

mssql-python 驅動程式支援 ? （qmark）與 %(name)s （pyformat）參數樣式。你現有的 ? 查詢不必變更也能運作：

變更前（pyodbc）：

cursor.execute("SELECT * FROM Production.Product WHERE ProductID = ? AND Color = ?", (1, "Red"))

接著（mssql-python，同一查詢）：

cursor.execute("SELECT * FROM Production.Product WHERE ProductID = ? AND Color = ?", (1, "Red"))

4. 保持 executemany 原樣

現有包含元組和 ? 標記的 executemany 呼叫無須變更即可運作：

之前（pyodbc）：

cursor.execute("CREATE TABLE #MigrateDemo (ID INT, Name NVARCHAR(50))")
data = [(1, "Alice"), (2, "Bob"), (3, "Carol")]
cursor.executemany("INSERT INTO #MigrateDemo (ID, Name) VALUES (?, ?)", data)

之後（mssql-python，相同程式碼）：

cursor.execute("DROP TABLE IF EXISTS #MigrateDemo")
cursor.execute("CREATE TABLE #MigrateDemo (ID INT, Name NVARCHAR(50))")
data = [(1, "Alice"), (2, "Bob"), (3, "Carol")]
cursor.executemany("INSERT INTO #MigrateDemo (ID, Name) VALUES (?, ?)", data)

預存程序移轉

mssql-python 驅動程式並未實作 callproc()。以下章節將說明如何使用 EXECUTE 替代。

對預存程序使用 EXECUTE

pyodbc 驅動程式支援，但 mssql-python 驅動程式不支援 callproc()。建議改用 EXECUTE ：

之前（pyodbc）：

cursor.callproc("dbo.uspGetEmployeeManagers", (5,))
results = cursor.fetchall()

在（mssql-python）之後：

cursor.execute(
    "EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = %(id)s",
    {"id": 5}
)
results = cursor.fetchall()
print(f"Got {len(results)} rows")

輸出參數

使用 T-SQL 變數擷取輸出值，而非依賴 callproc() 輸出參數：

之前（pyodbc，使用 callproc）：

params = (category_id, pyodbc.SQL_INTEGER)
cursor.callproc("dbo.GetProductCount", params)
count = params[1].value

之後（mssql-python，使用T-SQL變數）：

cursor.execute(
    """
    DECLARE @count INT;
    SELECT @count = COUNT(*) FROM Production.Product
    WHERE ProductSubcategoryID = %(cat_id)s;
    SELECT @count AS ProductCount;
    """,
    {"cat_id": 1}
)
product_count = cursor.fetchval()
print(f"Product count: {product_count}")

特定功能的遷移

以下章節將介紹 pyodbc 的特定功能及其 mssql-python 等價物。

連線字串

pyodbc 關鍵字	MSSQL-Python 關鍵字	Notes
`DRIVER={...}`	不需要	ODBC 驅動程式是內建的。
`SERVER=`	`Server=`	行為沒有改變。
`DATABASE=`	`Database=`	行為沒有改變。
`Trusted_Connection=`	`Trusted_Connection=`	行為沒有改變。
`UID=` / `PWD=`	`UID=` / `PWD=`	行為沒有改變。
`Authentication=`	`Authentication=`	接受相同的價值觀。

自動提交

兩個驅動程式的自動提交行為相同：

pyodbc:

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

MSSQL-python：

conn.autocommit = True

散裝內襯

為了加快大量 INSERT 批次的處理速度，pyodbc 使用者會設定 fast_executemany = True。 mssql-python 驅動程式已針對參數化批次的 executemany 最佳化，因此中等規模的插入作業不需要特殊旗標。對於大量資料載入，建議優先使用 bulkcopy()，因為它會透過批次複製通訊協定串流傳送資料列，而且比逐一執行個別的 INSERT 陳述式快得多。完整工作流程請參見使用批量複製。

pyodbc:

cursor.fast_executemany = True
cursor.executemany(query, data)

在（mssql-python）之後，使用適中的批次大小搭配 executemany：

cursor.execute("DROP TABLE IF EXISTS #BulkTarget")
cursor.execute("CREATE TABLE #BulkTarget (ID INT, Name NVARCHAR(50))")
data = [(i, f"Item {i}") for i in range(100)]
cursor.executemany("INSERT INTO #BulkTarget (ID, Name) VALUES (?, ?)", data)
conn.commit()

在（mssql-python）之後，大量載入首選使用 bulkcopy：

cursor.execute("IF OBJECT_ID('##BulkTarget') IS NOT NULL DROP TABLE ##BulkTarget")
cursor.execute("CREATE TABLE ##BulkTarget (ID INT, Name NVARCHAR(50))")
conn.commit()  # Commit DDL before bulkcopy
data = [(i, f"Item {i}") for i in range(100)]
result = cursor.bulkcopy("##BulkTarget", data)
print(f"Bulk copied {result['rows_copied']} rows")
cursor.execute("DROP TABLE ##BulkTarget")
conn.commit()

排廠

mssql-python 驅動程式預設會回傳支援屬性存取的 Row 物件，無需自訂資料列工廠：

pyodbc（自訂資料列工廠）：

def namedtuple_row_factory(cursor):
    from collections import namedtuple
    columns = [col[0] for col in cursor.description]
    Row = namedtuple("Row", columns)
    return Row

MSSQL-python（預設屬性存取權限）：

cursor.execute("SELECT Name, ListPrice FROM Production.Product")
row = cursor.fetchone()
print(row.Name)   # Attribute access works directly
print(row[0])     # Index access also works

錯誤處理

mssql-python 驅動程式使用與 pyodbc 相同的例外階層結構，因此大多數例外處理程序只需更改模組名稱即可。

例外狀況階層

例外類別名稱可在各驅動程式之間直接對應：

pyodbc：

try:
    cursor.execute(query)
except pyodbc.Error as e:
    pass
except pyodbc.DatabaseError as e:
    pass
except pyodbc.OperationalError as e:
    pass

MSSQL-python：

try:
    cursor.execute("SELECT TOP 1 * FROM Production.Product")
    print(cursor.fetchone())
except mssql_python.Error as e:
    pass
except mssql_python.DatabaseError as e:
    pass
except mssql_python.OperationalError as e:
    pass

錯誤詳細資料

兩個驅動程式都透過例外參數暴露錯誤細節：

pyodbc:

try:
    cursor.execute(query)
except pyodbc.Error as e:
    sqlstate = e.args[0]
    message = e.args[1]

MSSQL-python：

try:
    cursor.execute("SELECT TOP 1 * FROM NonExistentTable_XYZ")
except mssql_python.Error as e:
    # Error message contains SQLSTATE and details
    print(str(e))

連線池化

mssql-python 驅動程式預設包含連線池，因此不再需要外部池化函式庫。

移除外部池化

如果你在 pyodbc 中使用外部連線集區，mssql-python 驅動程式已內建此功能：

之前（pyodbc 外部連線池）：

from dbutils.pooled_db import PooledDB

pool = PooledDB(pyodbc, 5, driver="{ODBC Driver 18 for SQL Server}",
                server="your_server", database="your_database",
                uid="your_username", pwd="your_password")
conn = pool.connection()

之後（mssql-python 內建池化）：

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

設定集區

使用 mssql_python.pooling() 覆寫預設的集區大小和逾時設定：

import mssql_python

mssql_python.pooling()

完整遷移範例

以下展示用 pyodbc 寫的同一個函式，然後用 mssql-python 重寫。

變更前（pyodbc）

此版本使用 pyodbc 連接字串並搭配DRIVER關鍵字：

import pyodbc
from datetime import date

def get_orders(customer_id: int, start_date: date):
    conn = pyodbc.connect(
        "DRIVER={ODBC Driver 18 for SQL Server};"
        "SERVER=localhost;"
        "DATABASE=AdventureWorks2022;"
        "Trusted_Connection=yes;"
    )
    cursor = conn.cursor()

    cursor.execute("""
        SELECT SalesOrderID, OrderDate, TotalDue
        FROM Sales.SalesOrderHeader
        WHERE CustomerID = ? AND OrderDate >= ?
        ORDER BY OrderDate DESC
    """, (customer_id, start_date))

    orders = []
    for row in cursor:
        orders.append({
            "id": row.SalesOrderID,
            "date": row.OrderDate,
            "total": row.TotalDue
        })

    cursor.close()
    conn.close()
    return orders

之後（mssql-python）

此版本移除了關鍵字 DRIVER 。所有查詢、參數及列存取模式保持相同：

import mssql_python
from datetime import date

def get_orders(customer_id: int, start_date: date):
    conn = mssql_python.connect(
        "Server=localhost;"
        "Database=AdventureWorks2022;"
        "Trusted_Connection=yes;"
    )
    cursor = conn.cursor()

    cursor.execute("""
        SELECT SalesOrderID, OrderDate, TotalDue
        FROM Sales.SalesOrderHeader
        WHERE CustomerID = ? AND OrderDate >= ?
        ORDER BY OrderDate DESC
    """, (customer_id, start_date))

    orders = []
    for row in cursor:
        orders.append({
            "id": row.SalesOrderID,
            "date": row.OrderDate,
            "total": row.TotalDue
        })

    cursor.close()
    conn.close()
    return orders

唯一的變動是 import 語句和連接字串（不需要DRIVER關鍵字）。每個查詢、參數、擷取模式和列存取都保持不變。

遷移測試

在完成遷移前，先對兩個驅動執行相同的查詢並比較結果，以確認行為是否相同。

驗證等效行為

使用比較函式，對兩個驅動程式執行相同查詢，並斷言結果相符：

import pyodbc
import mssql_python

def compare_results(pyodbc_conn_str: str, mssql_conn_str: str, query: str):
    """Compare results from both drivers."""
    # pyodbc query
    pyodbc_conn = pyodbc.connect(pyodbc_conn_str)
    pyodbc_cursor = pyodbc_conn.cursor()
    pyodbc_cursor.execute(query)
    pyodbc_results = pyodbc_cursor.fetchall()
    pyodbc_conn.close()
    
    # mssql-python query
    mssql_conn = mssql_python.connect(mssql_conn_str)
    mssql_cursor = mssql_conn.cursor()
    mssql_cursor.execute(query)
    mssql_results = mssql_cursor.fetchall()
    mssql_conn.close()
    
    # Compare
    assert len(pyodbc_results) == len(mssql_results)
    for p_row, m_row in zip(pyodbc_results, mssql_results):
        assert tuple(p_row) == tuple(m_row)
    
    print(f"Results match: {len(pyodbc_results)} rows")

檢查清單

[ ] 將匯入來源從 pyodbc 更新為 mssql_python。
[ ] 從連接線中移除 DRIVER= 。
[ ] 保留現有的 ? 參數查詢（它們原樣即可運作）。
[ ] 使用 EXECUTE 語句來呼叫儲存程序。
[ ] 移除外部連線池設定。
[ ] 更新例外處理類別名稱。
[ ] 測試所有查詢與儲存程序。
[ ] 確認資料型別處理（尤其是小數點與日期）。
[ ] 將 ODBC 驅動程式從部署需求中移除。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22

從 pyodbc 遷移到 mssql-python

主要差異

基本遷移步驟

1. 更新匯入

2. 更新連接字串

3. 保持查詢原樣

4. 保持 executemany 原樣

預存程序移轉

對預存程序使用 EXECUTE

輸出參數

特定功能的遷移

連線字串

自動提交

散裝內襯

排廠

錯誤處理

例外狀況階層

錯誤詳細資料

連線池化

移除外部池化

設定集區

完整遷移範例

變更前（pyodbc）

之後（mssql-python）

遷移測試

驗證等效行為

檢查清單

相關內容

意見反應

其他資源