Migrar de pymssql 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 do FreeTDS.
• Vários cursores concorrentes por ligação.
• Agrupamento de ligações integrado.
• Suporte moderno para Python 3.10+.
• Autenticação nativa da Microsoft Entra.
• Objetos de linha com acesso a atributos por defeito.

Diferenças principais

Etapas básicas de migração

Os passos seguintes percorrem as alterações mais comuns necessárias para migrar uma aplicação pymssql para mssql-python.

1. Atualizar importações

Substitua a pymssql importação por mssql_python:

Antes (pymssql):

import pymssql

Depois (mssql-python):

import mssql_python

2. Atualizar chamadas de ligação

O pymssql utiliza argumentos posicionais. O driver mssql-python utiliza uma cadeia de ligação ou argumentos de palavras-chave:

Antes (pymssql, argumentos posicionais):

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

Antes (pymssql, argumentos-chave):

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

Depois (mssql-python, cadeia de ligação):

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

Depois (mssql-python, recomendado pela Microsoft Entra):

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

3. Atualizar marcadores de parâmetros

O pymssql utiliza %s e %d formata marcadores de posição. O driver mssql-python utiliza ? (qmark) ou %(name)s (pyformat):

Antes (pymssql):

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

Depois (mssql-python, estilo 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. Atualizar executemany

Atualize os marcadores SQL de %s/%d para ? ou :%(name)s

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

Depois (mssql-python, estilo 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. Usar atributos de linha em vez de as_dict cursores

O pymssql exige as_dict=True para aceder às colunas por nome. O driver mssql-python devolve Row objetos que suportam tanto o acesso a atributos como a índices por defeito:

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

Depois (mssql-python, acesso a atributos por defeito):

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]

Migração de procedimentos armazenados

O driver mssql-python não implementa callproc(). Utilize instruções EXECUTE em vez disso.

Utilize EXECUTE para procedimentos armazenados

PymsSQL suporta callproc(), mas o driver MSSQL-Python não. Use EXECUTE em vez disso:

Antes (pymssql):

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

Depois (mssql-python):

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

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

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

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

Migração de cópias em lote

O pymssql chama bulk_copy() na ligação. O driver mssql-python chama bulkcopy() no cursor com mais opções:

Antes (pymssql):

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

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

O método mssql-python bulkcopy() suporta batch_size, timeout, column_mappings, keep_identity, check_constraintstable_lock, , keep_nulls, fire_triggers, , e use_internal_transaction. Consulte cópia em massa para obter mais detalhes.

Vários cursores

O pymssql permite apenas um cursor ativo por ligação. O driver mssql-python suporta múltiplos cursores concorrentes:

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

Depois (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)}")

Agrupamento de conexões

O Pymssql não tem pooling incorporado. O driver mssql-python inclui-o automaticamente:

Antes (pymssql, é necessário um pool externo):

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

Depois (mssql-python, o pooling é automático):

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

Tratamento de erros

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

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

Depois (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}")

Exemplo completo de migração

O que se segue mostra a mesma função escrita com pymssql e depois reescrita com mssql-python.

Antes (pymssql)

Esta versão utiliza argumentos de conexão posicional, as_dict=True, e %d marcadores de parâmetros:

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

Depois (mssql-python)

As principais alterações estruturais são os marcadores de parâmetros, o estilo de conexão e o acesso por linha:

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

As alterações estruturais são:

1. import pymssql → import mssql_python.
2. Argumentos posicionais de ligação → cadeia de ligação com palavras-chave.
3. %d marcador de parâmetros → ?.
4. cursor(as_dict=True) → cursor() com acesso a atributos (row.SalesOrderID em vez de row["SalesOrderID"]).

Lista de Verificação

• [ ] Atualizar importações de pymssql para mssql_python.
• [ ] Converter as chamadas de conexão de argumentos posicionais para cadeias de conexão.
• [ ] Converter %s/%d marcadores de parâmetros para ? ou .%(name)s
• [ ] Utilize instruções EXECUTE para chamadas a procedimentos armazenados.
• [ ] Use o acesso a atributos Row em vez de cursores as_dict=True.
• [ ] Migrar conn.bulk_copy() para cursor.bulkcopy().
• [ ] Remover a configuração de agrupamento externo de ligações.
• [ ] Remover o FreeTDS dos requisitos de implementação.
• [ ] Atualizar os nomes das classes de tratamento de exceções.
• [ ] Testar todas as consultas e procedimentos armazenados.

Conteúdo relacionado

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