Microsoft Python-Treiber für SQL Server – mssql-python

mssql-python ist der Python-Treiber von Microsoft für SQL Server, Azure SQL-Datenbank, Azure SQL Managed Instance und SQL-Datenbank in Microsoft Fabric. Es verwendet Direct Database Connectivity (DDBC), sodass du dich verbinden kannst, ohne einen externen Treibermanager zu installieren. Der Treiber unterstützt Python 3.10 oder neuer, entspricht der Python Database API Specification 2.0 und fügt Python-freundliche Verbesserungen für die tägliche Entwicklung hinzu.

Auswählen des Startpunkts

• Um ein lokales SQL Server-Beispiel schnell zum Laufen zu bringen, beginnen Sie mit Quickstart: Verbinden Sie sich mit dem mssql-python-Treiber.
• Um sich mit passwortloser Authentifizierung mit Azure SQL zu verbinden, beginnen Sie mit Microsoft Entra-Authentifizierung und Verbindungszeichenfolgen.
• Um Daten interaktiv zu erforschen, beginnen Sie mit Connect from a Jupyter Notebook oder Rapid Prototyping.
• Um große Datenmengen effizient zu bewegen, wählen Sie Bulk-Kopieroperationen oder den Bulk-Copy-Quickstart.
• Um von einem anderen Treiber zu migrieren, gehe zu Migrate from pyodbc, Migrate from pymssql, Migrate from SQLite oder Migrate from PostgreSQL.

Produktionsbasisplan für Azure SQL

Nutzen Sie dieses Beispiel als Ausgangspunkt für eine produktionsorientierte Azure SQL-Verbindung. Es liest Konfigurationen aus der Umgebung, authentifiziert sich mit verwalteter Identität und aktiviert die Verschlüsselung des Tabular Data Stream (TDS) 8.0. Zudem legt es Zeitlimits für die Anmeldung und für einzelne Abfragen fest, versucht bei vorübergehenden Fehlern eine Wiederholung mit exponentiellem Backoff (eine neue Verbindung bei Verbindungsfehlern, dieselbe Verbindung bei Abfragefehlern wie Deadlocks), protokolliert Ergebnisse und nutzt Kontextmanager zur Freigabe von Ressourcen.

Die Schlüsselwörter ConnectRetryCount und ConnectRetryInterval in der Verbindungszeichenfolge aktivieren die SQL Server-Leerlaufverbindungsresilienz: Der Treiber stellt eine getrennte Leerlaufverbindung transparent wieder her. Das unterscheidet sich vom Anwendungs-Level-Retry in diesem Beispiel, bei dem eine Abfrage erneut ausprobiert wird, die mit einem vorübergehenden Fehler wie einem Deadlock oder Query-Timeout fehlschlägt. Die beiden ergänzen sich, also behalte beide.

import logging
import os
import time

import mssql_python

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("app")

# Transient errors that require a fresh connection to recover.
CONNECT_RETRY_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
})

# Transient errors that leave the connection usable, such as a deadlock victim
# or a query timeout, so retry on the same connection.
QUERY_RETRY_ERRORS = frozenset({
    "Serialization failure",
    "Timeout expired",
})


def connect_with_retry(conn_str: str, max_attempts: int = 3, login_timeout_s: int = 5) -> mssql_python.Connection:
    """Open a connection, retrying transient failures with exponential backoff."""
    for attempt in range(1, max_attempts + 1):
        try:
            conn = mssql_python.connect(
                conn_str,
                attrs_before={mssql_python.SQL_ATTR_LOGIN_TIMEOUT: login_timeout_s},
            )
            logger.info("connected on attempt %d/%d", attempt, max_attempts)
            return conn
        except mssql_python.OperationalError as exc:
            if exc.driver_error not in CONNECT_RETRY_ERRORS or attempt == max_attempts:
                logger.error("connect failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = 2 ** (attempt - 1)  # 1s, 2s, 4s
            logger.warning(
                "connect attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)


def execute_with_retry(
    conn: mssql_python.Connection,
    sql: str,
    *params,
    max_attempts: int = 3,
    query_timeout_s: int = 10,
) -> mssql_python.Cursor:
    """Run sql on an open connection and return the ready-to-fetch cursor.

    Retries errors that leave the connection usable so callers don't wrap each
    query in its own function. Pass query values as parameters. Retry only
    idempotent statements; wrap writes in an explicit transaction.
    """
    for attempt in range(1, max_attempts + 1):
        cursor = mssql_python.Cursor(conn, timeout=query_timeout_s)
        try:
            cursor.execute(sql, *params)
            if attempt > 1:
                logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return cursor
        except mssql_python.OperationalError as exc:
            cursor.close()
            if exc.driver_error not in QUERY_RETRY_ERRORS or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = 2 ** (attempt - 1)  # 1s, 2s, 4s
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise RuntimeError("unreachable: the retry loop exits by return or raise")


def main() -> None:
    # Read configuration from the environment; never hard-code secrets.
    server = os.environ["SQL_SERVER"]      # for example, myserver.database.windows.net
    database = os.environ["SQL_DATABASE"]  # for example, AdventureWorks
    client_id = os.getenv("AZURE_CLIENT_ID")  # set for a user-assigned managed identity

    # Authenticate with the workload's managed identity over TDS 8.0 encryption.
    # ConnectRetryCount/ConnectRetryInterval transparently reconnect a dropped
    # idle connection; they don't replay a failed query.
    conn_str = (
        f"Server={server};"
        f"Database={database};"
        "Authentication=ActiveDirectoryMsi;"
        "Encrypt=strict;"
        "ConnectRetryCount=3;"
        "ConnectRetryInterval=10;"
        # Parallel dials to all resolved IPs; safe on single-IP targets.
        "MultiSubnetFailover=Yes;"
    )
    if client_id:
        conn_str += f"UID={client_id};"

    query = """
        SELECT TOP 10
            p.BusinessEntityID,
            p.FirstName,
            p.LastName
        FROM Person.Person AS p
        ORDER BY p.BusinessEntityID;
    """

    try:
        # Context managers close the cursor and connection automatically.
        with connect_with_retry(conn_str) as conn:
            with execute_with_retry(conn, query) as cursor:
                for business_entity_id, first_name, last_name in cursor.fetchall():
                    print(f"{business_entity_id}\t{first_name}\t{last_name}")
    except mssql_python.Error:
        logger.exception("query failed")
        raise


if __name__ == "__main__":
    main()

Für ausführlichere Hinweise zu jedem Thema in diesem Beispiel siehe Microsoft Entra Authentifizierung, Connection Pooling, Verschlüsselung und Zertifikate, Retry-Logik und Fehlerbehandlung.

Wichtigste Funktionen

• PEP 249-Konformität: Standard connect, cursor, execute, und fetch* Schnittstellen sowie Pythonic-Erweiterungen.
• Direkte Datenbankverbindung (DDBC): Kein externer Treibermanager erforderlich. Installieren Sie mssql-python, und schon können Sie eine Verbindung herstellen.
• Microsoft Entra ID-Authentifizierung: Integrierte Unterstützung für Authentifizierungsmodi, einschließlich verwalteter Identitäten und Dienstprinzipalen.
• SQL Server und Windows-Authentifizierung: SQL-Logins, Kerberos und Windows Single Sign-on (SSO) auf unterstützten Plattformen.
• Bulk-Kopie: Hochleistungsfähiges Bulk-Einfügen für große Datenmengen mit nativer TDS-Protokollunterstützung.
• Native Unterstützung von Datentypen: JSON, XML, räumliche Datentypen, Sparse-Spalten, datetimeoffset und decimal/money mit präziser Verarbeitung.
• Apache Arrow-Integration: Zero-Copy-Ergebnissätze für schnellen Datenaustausch mit pandas, Polars und DuckDB.
• Asynchrone Muster: Verwenden Sie den Treiber mit asyncio-basierten Anwendungen und FastAPI mithilfe von Workarounds mit dem ThreadPoolExecutor. Siehe Asynkrone Muster für Integrationsmuster .
• TLS standardmäßig: TLS-Verschlüsselung und Zertifikatsvalidierung standardmäßig aktiviert (über ODBC Driver 18). TDS 8.0-Verschlüsselung ist verfügbar, wenn Sie setzen Encrypt=strict.

Loslegen

Konfigurieren und authentifizieren

Mit Daten arbeiten

SQL Server-Datentypen und -Funktionen

Integration mit Python-Tools und Frameworks

Bereitstellen und Betreiben

Migration zu mssql-python

Verwandte Inhalte

• Unterstützungslebenszyklus für mssql-python
• Was gibt es Neues in mssql-python
• mssql-python auf GitHub
• mssql-python on PyPI
• Veröffentlichungshinweise
• Roadmap
• Azure Python Entwicklerzentrum
• Python Database API Spezifikation 2.0