Migration von pyodbc zu mssql-python

Der mssql-python-Treiber ist Microsofts hauseigener Python-Treiber für Microsoft SQL. Wenn Sie eine von Microsoft getreue Treiberoption bevorzugen, bietet sie Folgendes an:

• Keine externe ODBC-Treiberabhängigkeit.
• Eingebautes Verbindungspooling.
• Moderne Unterstützung für Python 3.10+
• Native Microsoft Entra-Authentifizierung.

Wichtige Unterschiede

Grundlegende Migrationsschritte

Die folgenden Schritte behandeln die Schlüsseländerungen, um eine Pyodbc-Anwendung auf mssql-python zu migrieren.

1. Importe aktualisieren

Ersetzen Sie den Import pyodbc durch mssql_python:

Vorher (pyodbc):

import pyodbc

Nach (mssql-python):

import mssql_python

2. Verbindungsstrings aktualisieren

Entfernen Sie das DRIVER= Schlüsselwort und aktualisieren Sie die Authentifizierungsmethode:

Vorher (pyodbc, benötigt ODBC-Treiber):

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

Danach (mssql-python, kein Treiber erforderlich, verwendet Microsoft Entra-Authentifizierung):

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

3. Halten Sie Ihre Anfragen unverändert

Der mssql-python-Treiber unterstützt sowohl ? (qmark) als auch %(name)s (pyformat) Parameterstile. Ihre bestehenden ? Abfragen funktionieren ohne Änderungen:

Vorher (pyodbc):

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

Danach (mssql-python, gleiche Abfrage):

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

4. Executemany unverändert beibehalten

Bestehende executemanyAufrufe mit Tupeln und ?-Markierungen funktionieren ohne Änderungen:

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

Danach (mssql-python, gleicher Code):

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)

Migration von gespeicherten Prozeduren

Der mssql-python-Treiber implementiert callproc()nicht . Die folgenden Abschnitte zeigen, wie man stattdessen verwendet EXECUTE .

Verwenden Sie EXECUTE für gespeicherte Prozeduren

Der pyodbc-Treiber unterstützt callproc(), aber der mssql-python-treiber nicht. Verwenden Sie EXECUTE stattdessen:

Vorher (pyodbc):

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

Nach (mssql-python):

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

Ausgabeparameter

Verwenden Sie T-SQL-Variablen, um Ausgabewerte zu erfassen, anstatt sich auf callproc() Ausgabeparameter zu verlassen:

Vorher (pyodbc, mit callproc):

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

Danach (mssql-python, unter Verwendung von T-SQL-Variablen):

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

Merkmalsspezifische Migrationen

Die folgenden Abschnitte behandeln spezifische pyodbc-Funktionen und deren mssql-python-Äquivalente.

Verbindungszeichenfolgen

Autocommit

Das Autocommit-Verhalten ist bei beiden Treibern identisch:

Pyodbc:

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

mssql-python:

conn.autocommit = True

Schüttguteinsätze

Um große INSERT Chargen zu beschleunigen, setzen pyodbc-Nutzer fast_executemany = True. Der mssql-python-Treiber optimiert executemany bereits für parametrisierte Batches, daher benötigen moderate Einfügungen kein spezielles Flag. Für große Datenmengen verwenden Sie vorzugsweise bulkcopy(), da hiermit Zeilen über das Bulk-Copy-Protokoll gestreamt werden und dies viel schneller ist als das einzelne Ausführen von INSERT-Anweisungen. Für den vollständigen Workflow siehe Massekopie verwenden.

Pyodbc:

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

Nach (mssql-python) moderate Batches mit 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()

Nach (mssql-python), große Datenlasten mit bulkcopy (bevorzugt):

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

Reihenfabrik

Der mssql-python-Treiber gibt standardmäßig RowObjekte zurück, die den Attributzugriff unterstützen, ohne dass eine benutzerdefinierte Row-Factory erforderlich ist:

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 (standardmäßig attributzugriff):

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

Fehlerbehandlung

Der mssql-python-Treiber verwendet dieselbe Ausnahmehierarchie wie pyodbc, daher benötigen die meisten Ausnahmehandler nur eine Modulnamensänderung.

Ausnahmehierarchie

Die Namen der Ausnahmeklassen werden direkt zwischen den Treibern abgebildet:

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

Fehlerdetails

Beide Treiber liefern Fehlerdetails durch Ausnahmeargumente:

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

Verbindungspooling

Der mssql-python-Treiber enthält standardmäßig Connection Pooling, sodass externe Pooling-Bibliotheken nicht mehr benötigt werden.

Externes Pooling entfernen

Wenn du externes Pooling mit pyodbc verwendet hast, hat der mssql-python-Treiber das eingebaut:

Vor (externer pyodbc-Pool):

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

Nach (mssql-python integriertes Pooling):

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

Pool konfigurieren

Überschreiben Sie die Standard-Poolgröße und den Timeout-Wert mit mssql_python.pooling():

import mssql_python

mssql_python.pooling()

Vollständiges Migrationsbeispiel

Das Folgende zeigt dieselbe Funktion, die mit pyodbc geschrieben und dann mit mssql-python neu geschrieben wurde.

Vorher (pyodbc)

Diese Version verwendet den pyodbc-Verbindungszeichenfolge mit einem DRIVER Schlüsselwort:

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

Danach (mssql-python)

Diese Version entfernt das DRIVER Schlüsselwort. Alle Abfragen, Parameter und Zeilenzugriffsmuster bleiben identisch:

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

Die einzigen Änderungen sind die Import-Anweisung und die Verbindungszeichenfolge (kein DRIVER Schlüsselwort erforderlich). Jede Abfrage, jeder Parameter, jedes Abrufmuster und jeder Zeilenzugriff bleibt identisch.

Testen der Migration

Führen Sie vor Abschluss der Migration dieselben Abfragen bei beiden Fahrern durch und vergleichen Sie die Ergebnisse, um gleichwertiges Verhalten zu bestätigen.

Überprüfen Sie das äquivalente Verhalten

Verwenden Sie eine Vergleichsfunktion, die dieselbe Abfrage gegen beide Treiber ausführt und die Ergebnisübereinstimmung bestätigt:

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

• [ ] Importe aktualisieren von pyodbc zu mssql_python.
• [ ] Entfernen Sie DRIVER= aus den Verbindungszeichenfolgen.
• [ ] Behalten Sie bestehende ? Parameterabfragen (sie funktionieren as-is).
• [ ] Verwenden Sie EXECUTE Anweisungen für gespeicherte Prozeduraufrufe.
• [ ] Externe Verbindungspooling-Konfiguration entfernen.
• [ ] Aktualisieren der Klassennamen für Ausnahmebehandlung.
• [ ] Teste alle Abfragen und gespeicherten Verfahren.
• [ ] Überprüfen Sie die Verarbeitung des Datentyps (insbesondere Dezimalzahlen und Daten).
• [ ] Entfernen Sie den ODBC-Treiber aus den Einsatzanforderungen.

Verwandte Inhalte

• Migration von pymssql
• Installation
• Verbindungszeichenfolgen
• Abfragen ausführen
• Gespeicherten Prozeduren