Migrar de pyodbc para mssql-python

O controlador mssql-python é o controlador Python oficial da Microsoft para o Microsoft SQL. Se preferir uma opção de driver mantido pela Microsoft, oferece:

• Sem dependência de drivers ODBC externos.
• Agrupamento de ligações integrado.
• Suporte moderno para Python 3.10+.
• Autenticação nativa da Microsoft Entra.

Diferenças principais

Etapas básicas de migração

Os passos seguintes cobrem as principais alterações para migrar uma aplicação pyodbc para mssql-python.

1. Atualizar importações

Substitua a pyodbc importação por mssql_python:

Antes (pyodbc):

import pyodbc

Depois (mssql-python):

import mssql_python

2. Atualizar cadeias de ligação

Remover a DRIVER= palavra-chave e atualizar o método de autenticação:

Antes (pyodbc, requer o driver ODBC):

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

Depois (mssql-python, sem necessidade de driver, usando autenticação do Microsoft Entra):

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

3. Mantenha as suas consultas tal como estão

O driver mssql-python suporta os estilos de parâmetros ? (qmark) e %(name)s (pyformat). As suas consultas existentes ? funcionam sem alterações:

Antes (pyodbc):

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

Depois (mssql-python, mesma consulta):

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

4. Mantenha executemany como está

As chamadas existentes executemany com tuplos e marcadores ? continuam a funcionar sem alterações:

Antes (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)

Depois (mssql-python, mesmo código):

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)

Migração de procedimentos armazenados

O driver mssql-python não implementa callproc(). As secções seguintes mostram como usar EXECUTE em vez disso.

Utilize EXECUTE para procedimentos armazenados

O driver pyodbc suporta callproc(), mas o driver mssql-python não. Use EXECUTE em vez disso:

Antes (pyodbc):

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

Depois (mssql-python):

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

Parâmetros de saída

Use variáveis T-SQL para capturar valores de saída em vez de depender de callproc() parâmetros de saída:

Antes (pyodbc, usando callproc):

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

Depois (mssql-python, usando variáveis 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}")

Migrações específicas de funcionalidades

As secções seguintes cobrem funcionalidades específicas do pyodbc e os seus equivalentes mssql-python.

Cadeias de ligação

Confirmação automática

O comportamento do autocommit é idêntico em ambos os drivers:

pyodbc:

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

mssql-python:

conn.autocommit = True

Inserções em massa

Para acelerar grandes lotes de INSERT, os utilizadores de pyodbc configuram fast_executemany = True. O driver mssql-python já otimiza executemany para lotes parametrizados, por isso inserções moderadas não precisam de flag especial. Para cargas de dados grandes, prefira bulkcopy(), que transmite linhas através do protocolo de cópia em massa e é muito mais rápido do que emitir instruções individuais INSERT . Para o fluxo de trabalho completo, veja Usar cópia em massa.

pyodbc:

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

Após (mssql-python), faz lotes moderados com 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()

Após (mssql-python), cargas de grande volume com bulkcopy (preferencial):

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()

Fábrica de linhas

O controlador mssql-python retorna objetos Row que permitem o acesso a atributos por predefinição, sem exigir uma função de criação de linhas personalizada:

PYODBC (Custom Row Factory):

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 (acesso a atributos por defeito):

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

Tratamento de erros

O driver mssql-python usa a mesma hierarquia de exceções do pyodbc, pelo que a maioria dos tratadores de exceções requer apenas a alteração do nome do módulo.

Hierarquia de exceção

Os nomes das classes de exceção são mapeados diretamente entre os drivers:

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

Detalhes do erro

Ambos os controladores expõem detalhes de erro através dos argumentos da exceção:

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))

Agrupamento de conexões

O controlador mssql-python inclui agrupamento de ligações por predefinição, pelo que as bibliotecas externas para agrupamento de ligações já não são necessárias.

Remover a agregação externa

Se usaste pooling externo com pyodbc, o driver mssql-python tem isso incorporado:

Antes (piscina externa 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()

Após (com pool de ligações incorporado no mssql-python):

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

Configurar agrupamento

Redefina o tamanho predefinido do pool e o timeout com mssql_python.pooling():

import mssql_python

mssql_python.pooling()

Exemplo completo de migração

O seguinte mostra a mesma função escrita com pyodbc e depois reescrita com mssql-python.

Antes (pyodbc)

Esta versão utiliza a cadeia de ligação pyodbc com uma palavra-chave 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

Depois (mssql-python)

Esta versão remove a DRIVER palavra-chave. Todas as consultas, parâmetros e padrões de acesso à linha permanecem idênticos:

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

As únicas alterações são a instrução de importação e a cadeia de ligação (sem DRIVER necessidade de palavra-chave). Cada consulta, parâmetro, padrão de obtenção e acesso a linhas permanece idêntico.

Teste a migração

Antes de concluir a migração, execute as mesmas consultas em ambos os drivers e compare os resultados para confirmar o comportamento equivalente.

Verificar comportamento equivalente

Use uma função de comparação que execute a mesma consulta contra ambos os drivers e afirme que os resultados coincidem:

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")

Lista de Verificação

• [ ] Atualizar importações de pyodbc para mssql_python.
• [ ] Retirar DRIVER= dos fios de ligação.
• [ ] Mantém as consultas de parâmetros existentes ? (funcionam as-is).
• [ ] Utilize instruções EXECUTE para chamadas a procedimentos armazenados.
• [ ] Remover a configuração de agrupamento externo de ligações.
• [ ] Atualizar os nomes das classes de tratamento de exceções.
• [ ] Testar todas as consultas e procedimentos armazenados.
• [ ] Verificar o tratamento dos tipos de dados (especialmente decimais e datas).
• [ ] Remover o driver ODBC dos requisitos de implementação.

Conteúdo relacionado

• Migrar de pymssql
• Installation
• Cadeias de ligação
• Execução de consultas
• Procedimentos armazenados