从 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()`	支持	未实现
`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）

清单

相关内容

反馈

其他资源

之后（mssql-python）