從 pymssql 遷移到 mssql-python

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

沒有 FreeTDS 依賴。
每個連線有多個並行游標。
內建連線池。
現代 Python 3.10+ 支援。
原生 Microsoft Entra 驗證。
預設列物件具有屬性存取權。

主要差異

Feature	PymsSQL	MSSQL-Python
參數樣式	`format`（`%s`，`%d`）	`qmark` （`?`）以及 `pyformat` （`%(name)s`）
原生圖書館	FreeTDS	DDBC（隨附）
連線池化	外部	內建
每個連線的游標數	1	Multiple
最低限度的 Python	3.6	3.10
`callproc()`	Supported	未實作
`as_dict` 游標	Extension	列物件（預設）
大量複製	`conn.bulk_copy()`	`cursor.bulkcopy()`
自動提交預設	Off	Off

基本遷移步驟

以下步驟將逐步介紹將 pymssql 應用程式遷移到 mssql-python 所需的最常見變更。

1. 更新匯入

將 pymssql 匯入替換為 mssql_python：

之前（pymssql）：

import pymssql

之後（mssql-python）：

import mssql_python

2. 更新連線呼叫

pymssql 使用位置參數。 mssql-python 驅動程式使用連線字串或關鍵字參數：

之前（pymssql，位置參數）：

conn = pymssql.connect("<server>", "<user>", "<password>", "<database>")

之前（pymssql，關鍵字參數）：

conn = pymssql.connect(
    host=r"<server>\<instance>",
    user="<login>",
    password="<password>",
    database="<database>"
)

在（mssql-python、連線字串）之後：

conn = mssql_python.connect(
    "Server=<server>;"
    "Database=<database>;"
    "UID=<username>;"
    "PWD=<password>;"
    "Encrypt=yes;"
)

之後（mssql-python，建議使用 Microsoft Entra）：

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

3. 更新參數標記

pymssql 使用 %s 和 %d 格式預留位置符。 mssql-python 驅動程式使用 ? （qmark）或 %(name)s （pyformat）：

之前（pymssql）：

cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = %d AND FirstName = %s", (user_id, name))

之後（mssql-python，qmark 樣式）：

user_id, name = 1, "Ken"
cursor.execute("SELECT * FROM Person.Person WHERE BusinessEntityID = ? AND FirstName = ?", (user_id, name))
print(cursor.fetchone())

cursor.execute(
    "SELECT * FROM Person.Person WHERE BusinessEntityID = %(id)s AND FirstName = %(name)s",
    {"id": user_id, "name": name}
)
print(cursor.fetchone())

4. 更新 executemany

將 SQL 佔位符從 %s/%d 或 ?%(name)s更新：

之前（pymssql）：

cursor.execute("CREATE TABLE #Persons (ID INT, Name NVARCHAR(50), Department NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Persons VALUES (%d, %s, %s)",
    [(1, "John", "Sales"), (2, "Jane", "Marketing")]
)

之後（mssql-python，qmark 樣式）：

cursor.execute("IF OBJECT_ID('#Persons') IS NOT NULL DROP TABLE #Persons")
cursor.execute("CREATE TABLE #Persons (ID INT, Name NVARCHAR(50), Department NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Persons VALUES (?, ?, ?)",
    [(1, "John", "Sales"), (2, "Jane", "Marketing")]
)
cursor.execute("SELECT * FROM #Persons")
for row in cursor:
    print(row)

5. 使用列屬性代替as_dict游標

pymssql 需要 as_dict=True 才能依名稱存取資料行。 mssql-python 驅動程式預設會傳回可同時透過屬性和索引存取的 Row 物件：

之前（pymssql）：

cursor = conn.cursor(as_dict=True)
cursor.execute("SELECT BusinessEntityID, FirstName FROM Person.Person WHERE FirstName = %s", ("John",))
for row in cursor:
    print("ID=%d, Name=%s" % (row["BusinessEntityID"], row["FirstName"]))

之後（mssql-python，預設屬性存取權限）：

cursor = conn.cursor()
cursor.execute("SELECT BusinessEntityID, FirstName FROM Person.Person WHERE FirstName = ?", ("John",))
for row in cursor:
    print(f"ID={row.BusinessEntityID}, Name={row.FirstName}")
    # Index access also works: row[0], row[1]

預存程序移轉

mssql-python 驅動程式並未實作 callproc()。改用 EXECUTE 語句。

對預存程序使用 EXECUTE

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

之前（pymssql）：

cursor.callproc("uspGetEmployeeManagers", (5,))
for row in cursor:
    print(row)

之後（mssql-python）：

cursor.execute("EXECUTE dbo.uspGetEmployeeManagers @BusinessEntityID = ?", (5,))
for row in cursor:
    print(row)

輸出參數

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

之前（pymssql）：

cursor.callproc("GetProductCount", (category_id,))
count = cursor.fetchval()

之後（mssql-python、T-SQL 變數）：

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

批量複製遷移

pymssql 在連線上呼叫 bulk_copy()。 mssql-python 驅動程式在游標上呼叫 bulkcopy() ，並提供更多選項：

之前（pymssql）：

conn.bulk_copy("##BulkDemo", [(1, 2)] * 1000)
conn.commit()

在（mssql-python）之後：

cursor = conn.cursor()
cursor.execute("CREATE TABLE ##BulkDemo (Col1 INT, Col2 INT)")
conn.commit()
result = cursor.bulkcopy("##BulkDemo", [(1, 2)] * 1000)
print(f"Copied {result['rows_copied']} rows")
conn.commit()
cursor.execute("DROP TABLE ##BulkDemo")
conn.commit()

mssql-python bulkcopy() 方法支援 batch_size、timeout、column_mappings、keep_identity、check_constraints、table_lock、keep_nulls、fire_triggers，以及 use_internal_transaction。詳情請參閱「批量副本」。

多個游標

pymssql 每個連線只允許一個主動游標。 mssql-python 驅動程式支援多個並行游標：

之前（pymssql）：

c1 = conn.cursor()
c1.execute("SELECT TOP 5 * FROM Person.Person")
c2 = conn.cursor()
c2.execute("SELECT TOP 5 * FROM Sales.SalesOrderHeader")
c1.fetchall()

之後（mssql-python）：

c1 = conn.cursor()
c1.execute("SELECT TOP 5 BusinessEntityID, FirstName FROM Person.Person")
persons = c1.fetchall()

c2 = conn.cursor()
c2.execute("SELECT TOP 5 SalesOrderID FROM Sales.SalesOrderHeader")
orders = c2.fetchall()
print(f"Persons: {len(persons)}, Orders: {len(orders)}")

連線池化

pymssql 沒有內建的池化功能。 mssql-python 驅動程式會自動包含它：

之前（pymssql，需外部池）：

from dbutils.pooled_db import PooledDB
pool = PooledDB(pymssql, host="server", user="user", password="pwd", database="db")
conn = pool.connection()

在（mssql-python）之後，池化是自動的：

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

錯誤處理

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

之前（pymssql）：

try:
    cursor.execute(query)
except pymssql.OperationalError as e:
    print(f"Operation failed: {e}")
except pymssql.InterfaceError as e:
    print(f"Interface error: {e}")

在（mssql-python）之後：

try:
    cursor = conn.cursor()
    cursor.execute("SELECT TOP 1 * FROM Production.Product")
    row = cursor.fetchone()
    print(row)
except mssql_python.OperationalError as e:
    print(f"Operation failed: {e}")
except mssql_python.InterfaceError as e:
    print(f"Interface error: {e}")

完整遷移範例

以下展示用 pymssql 寫成並用 mssql-python 重寫的同一函式。

之前（pymssql）

此版本使用位置連接參數， as_dict=True以及 %d 參數標記：

import pymssql

def get_orders(customer_id: int):
    conn = pymssql.connect("<server>", "<user>", "<password>", "<database>")
    cursor = conn.cursor(as_dict=True)

    cursor.execute("""
        SELECT TOP 10 SalesOrderID, OrderDate, TotalDue
        FROM Sales.SalesOrderHeader
        WHERE CustomerID = %d
        ORDER BY OrderDate DESC
    """, (customer_id,))

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

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

之後（mssql-python）

主要結構變更包括參數標記、連接樣式及列存取：

import mssql_python

def get_orders(customer_id: int):
    conn = mssql_python.connect(
        "Server=<server>;"
        "Database=<database>;"
        "UID=<username>;"
        "PWD=<password>;"
    )
    cursor = conn.cursor()

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

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

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

結構上的變動包括：

import pymssql → import mssql_python。
位置式連線引數 → 含關鍵字的連線字串。
%d 參數標記→ ?。
cursor(as_dict=True) → cursor() 屬性存取權（row.SalesOrderID 而非 row["SalesOrderID"]）。

檢查清單

[ ] 將匯入來源從 pymssql 更新為 mssql_python。
[ ] 將連接呼叫從位置參數轉換為連接字串。
[ ] 將參數標記%s轉換為/%d?或。%(name)s
[ ] 使用 EXECUTE 語句來呼叫儲存程序。
[ ] 使用 Row 屬性存取代替 as_dict=True 游標。
[ ] 遷移 conn.bulk_copy() 到 cursor.bulkcopy()。
[ ] 移除外部連線池設定。
[ ] 將 FreeTDS 從部署要求中移除。
[ ] 更新例外處理類別名稱。
[ ] 測試所有查詢與儲存程序。

意見反應

此頁面對您有幫助嗎？

Last updated on 2026-07-22

從 pymssql 遷移到 mssql-python

主要差異

基本遷移步驟

1. 更新匯入

2. 更新連線呼叫

3. 更新參數標記

4. 更新 executemany

5. 使用列屬性代替as_dict游標

預存程序移轉

對預存程序使用 EXECUTE

輸出參數

批量複製遷移

多個游標

連線池化

錯誤處理

完整遷移範例

之前（pymssql）

之後（mssql-python）

檢查清單

相關內容

意見反應

其他資源