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 sich mit SQL Server, Azure SQL-Datenbank oder SQL-Datenbank in Microsoft Fabric zu verbinden und Ihre erste Abfrage auszuführen, beginnen Sie mit Quickstart: Verbinden Sie sich mit dem mssql-python-Treiber.
• Um einen lokalen SQL Server-Container zu erstellen, beginne mit Container- und lokaler Entwicklung.
• 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

Dieses Beispiel verbindet sich mit Azure SQL mit verwalteter Identität und Tabular Data Stream (TDS) 8.0-Verschlüsselung. Es liest die Verbindungseinstellungen aus Umgebungsvariablen und legt separate Anmelde- und Abfragezeitlimits fest. Jeder erneute Versuch ruft connect() erneut auf, anstatt ein fehlgeschlagenes Verbindungsobjekt wiederzuverwenden. Context Manager geben die Verbindung und den Cursor nach jedem Versuch frei.

Verwenden Sie diesen Helfer nur für idempotente Leseanfragen, die sicher wiederholt werden können. Nutze es nicht, um einzelne Schreibvorgänge oder teilweise abgeschlossene Transaktionen erneut abzuspielen.

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.

Die Azure-Fehlerprüfung stimmt mit der Standard-englischen Fehlermeldung 40501 überein, selbst wenn die Diagnose die Fehlernummer weglässt. Es prüft ddbc_error auch auf bekannte Zahlen, wenn vorhanden. Matching ist Best-Effort: Lokalisierte oder anders formatierte Nachrichten passen möglicherweise nicht überein. Weitere Informationen finden Sie unter Vorübergehende Verbindungsfehler.

import logging
import os
import re
import time

import mssql_python

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

TRANSIENT_DRIVER_ERRORS = frozenset({
    "Timeout expired",
    "Connection timeout expired",
    "Client unable to establish connection",
    "Communication link failure",
    "Connection failure during transaction",
    "Serialization failure",
    "Statement completion unknown",
})

AZURE_THROTTLING_ERRORS = frozenset({
    40197, 40501, 40540, 40613, 40680, 49918, 49919, 49920, 10928, 10929,
})

_ERROR_NUMBER_RE = re.compile(r"\b(?:Error|Msg)\s+(\d+)\b")
_AZURE_40501_MESSAGE = "The service is currently busy. Retry the request after 10 seconds."


def is_azure_throttling(error: BaseException) -> bool:
    """Best-effort detection of Azure SQL throttling in ProgrammingError text."""
    if not isinstance(error, mssql_python.ProgrammingError):
        return False
    ddbc_text = getattr(error, "ddbc_error", "") or ""
    return (
        _AZURE_40501_MESSAGE in ddbc_text
        or any(int(m) in AZURE_THROTTLING_ERRORS for m in _ERROR_NUMBER_RE.findall(ddbc_text))
    )


def query_with_retry(
    conn_str: str,
    sql: str,
    *params,
    max_attempts: int = 3,
    login_timeout_s: int = 5,
    query_timeout_s: int = 10,
) -> list:
    """Fetch an idempotent read query, reconnecting before each retry."""
    for attempt in range(1, max_attempts + 1):
        try:
            with mssql_python.connect(conn_str, timeout=login_timeout_s) as conn:
                conn.timeout = query_timeout_s
                with conn.cursor() as cursor:
                    cursor.execute(sql, *params)
                    rows = cursor.fetchall()
            logger.info("query succeeded on attempt %d/%d", attempt, max_attempts)
            return rows
        except mssql_python.Error as exc:
            azure_throttling = is_azure_throttling(exc)
            transient = (
                isinstance(exc, mssql_python.OperationalError)
                and exc.driver_error in TRANSIENT_DRIVER_ERRORS
            )
            if not (transient or azure_throttling) or attempt == max_attempts:
                logger.error("query failed on attempt %d/%d: %s", attempt, max_attempts, exc.driver_error)
                raise
            delay = (10 if azure_throttling else 1) * (2 ** (attempt - 1))
            logger.warning(
                "query attempt %d/%d hit transient error %r; retrying in %ds",
                attempt, max_attempts, exc.driver_error, delay,
            )
            time.sleep(delay)
    raise ValueError("max_attempts must be at least 1")


def main() -> None:
    # Read configuration from the environment; never hard-code secrets.
    server = os.environ["SQL_SERVER"]      # for example, <server>.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:
        rows = query_with_retry(conn_str, query)
        for business_entity_id, first_name, last_name in rows:
            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 mit einem der beiden nativen ODBC-Anbieter aktiviert. 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

Anfordern eines Features

Um eine Funktion anzufordern, öffne ein Problem im mssql-python GitHub-Repository.

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