Migrar de pyodbc para mssql-python

O driver mssql-python é o driver Python de primeira parte da Microsoft para Microsoft SQL. Se você preferir uma opção de driver mantida pela Microsoft, ela oferece:

  • Sem dependência de driver ODBC externo.
  • Pool de conexões integrado.
  • Suporte moderno para Python 3.10+.
  • Autenticação nativa do Microsoft Entra.

Principais diferenças

Característica pyodbc mssql-python
Estilo de parâmetros qmark (?) qmark (?) e pyformat (%(name)s)
É necessário um driver ODBC Sim No
Agrupamento de conexões Externo Incorporado
Versão mínima do Python 3,6 3.10
callproc() Supported Não implementado
Autocommit padrão Off Off

Etapas básicas de migração

Os passos a seguir cobrem as principais mudanças para migrar uma aplicação pyodbc para mssql-python.

1. Atualize as importações

Substitua a pyodbc importação por mssql_python:

Antes (pyodbc):

import pyodbc

Depois (mssql-python):

import mssql_python

2. Atualizar cadeias de conexão

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

Antes (pyodbc, requer 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 suas consultas como estão

O driver mssql-python oferece suporte aos estilos de parâmetro ? (qmark) e %(name)s (pyformat). 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á

Chamadas executemany existentes com tuplas e marcadores ? funcionam 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 seções seguintes mostram como usar EXECUTE em vez disso.

Use EXECUTE para procedimentos armazenados

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

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 por funcionalidade

As seções seguintes abordam características específicas do pyodbc e seus equivalentes mssql-python.

Cadeias de conexão

Palavra-chave pyodbc Palavra-chave mssql-python Notes
DRIVER={...} Não é necessário O driver ODBC é incluído internamente.
SERVER= Server= Nenhuma alteração de comportamento.
DATABASE= Database= Nenhuma alteração de comportamento.
Trusted_Connection= Trusted_Connection= Nenhuma alteração de comportamento.
UID= / PWD= UID= / PWD= Nenhuma alteração de comportamento.
Authentication= Authentication= Aceita os mesmos valores.

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 INSERT grandes lotes, usuários de pyodbc configuram fast_executemany = True. O driver mssql-python já otimiza executemany para lotes parametrizados, então inserções moderadas não precisam de flag especial. Para cargas de dados grandes, prefira bulkcopy(), que transmite linhas pelo 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), faça 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 grandes com bulkcopy (preferido):

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 driver mssql-python retorna objetos Row que permitem acesso por atributo por padrão, sem exigir uma row factory personalizada:

PYODBC (fábrica de fileiras personalizadas):

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 padrão):

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, então a maioria dos manipuladores de exceções requer apenas a mudança do nome do módulo.

Hierarquia de exceções

Os nomes das classes de exceção mapeiam 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 drivers expõem detalhes de erro por meio de argumentos de 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 driver mssql-python inclui pooling de conexões por padrão, então bibliotecas externas de pooling não são mais necessárias.

Remover agrupamento externo

Se você usou pooling externo com pyodbc, o driver mssql-python já tem isso embutido:

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

Depois (pooling interno do mssql-python):

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

Configurar pool

Substitua os valores padrão do tamanho do pool e do tempo limite usando mssql_python.pooling():

import mssql_python

mssql_python.pooling()

Exemplo de migração completa

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

Antes (pyodbc)

Esta versão usa a cadeia de conexão do 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 mudanças são a instrução de importação e a cadeia de conexão (sem necessidade da palavra-chave DRIVER). Cada consulta, parâmetro, padrão de busca e acesso a linhas permanece idêntico.

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

Checklist

  • [ ] Atualizar importações de pyodbc para mssql_python.
  • [ ] Remova DRIVER= das strings de conexão.
  • [ ] Mantenha consultas de parâmetros existentes ? (elas funcionam as-is).
  • [ ] Use instruções EXECUTE para chamadas a procedimento armazenado.
  • [ ] Remover a configuração de pool de conexão externa.
  • [ ] Atualize os nomes das classes de tratamento de exceções.
  • [ ] Teste todas as consultas e procedimentos armazenados.
  • [ ] Verificar o manuseio dos tipos de dados (especialmente decimais e datas).
  • [ ] Remover o driver ODBC dos requisitos de implantação.