从 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驱动	是的	否
连接池	外部	内置
最低限度的Python	3.6	3.10
`callproc()`	支持	未实现
自动提交默认	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 关键字	注释
`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）

该版本使用包含 DRIVER 关键字的 pyodbc 连接字符串：

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

唯一的变化是导入语句和连接字符串（不需要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）

迁移测试

验证等效行为

清单

相关内容

反馈

其他资源

之后（mssql-python）