Microsoft Python-stuurprogramma voor SQL Server - mssql-python

mssql-python is het Python-stuurprogramma van Microsoft voor SQL Server, Azure SQL Database, Azure SQL Managed Instance en de SQL-database in Microsoft Fabric. Het gebruikt Direct Database Connectivity (DDBC), zodat je kunt verbinden zonder een externe driver manager te installeren. De driver ondersteunt Python 3.10 of later en voldoet aan de Python Database API Specification 2.0, terwijl er Python-vriendelijke verbeteringen worden toegevoegd voor dagelijkse ontwikkeling.

Uw beginpunt kiezen

• Om snel een lokaal SQL Server-voorbeeld te laten werken, begin je met Quickstart: Verbind met de mssql-python-driver.
• Om verbinding te maken met Azure SQL met wachtwoordloze authenticatie, begin je met Microsoft Entra-authenticatie- en Connection-strings.
• Om data interactief te verkennen, begin je met Verbinden vanuit een Jupyter-notebook of Snel prototypen.
• Om grote hoeveelheden data efficiënt te verplaatsen, ga je naar Bulk copy operations of de Bulk copy quickstart.
• Om te migreren van een andere driver, ga je naar Migreren van pyodbc, Migreren van pymssql, migreren van SQLite, of migreren vanaf PostgreSQL.

Productiebasislijn voor Azure SQL

Gebruik dit voorbeeld als uitgangspunt voor een productiegerichte Azure SQL-verbinding. Het leest configuraties uit de omgeving, authenticeert met beheerde identiteit en schakelt Tabular Data Stream (TDS) 8.0-encryptie in. Het stelt ook time-outs voor aanmelden en per statement/query in, probeert tijdelijke fouten opnieuw met exponentiële back-off (een nieuwe verbinding bij verbindingsfouten, dezelfde verbinding bij queryfouten zoals deadlocks), registreert de resultaten en vertrouwt op contextmanagers om resources vrij te geven.

De ConnectRetryCount en ConnectRetryInterval trefwoorden in de verbindingsreeks maken de veerkracht van de idle-verbinding van SQL Server mogelijk: de driver maakt transparant opnieuw verbinding met een verbroken idle-verbinding. Dat verschilt van de herhalingspoging op toepassingsniveau in dit voorbeeld, waarbij een query opnieuw wordt uitgevoerd als die mislukt door een tijdelijke fout, zoals een deadlock of een time-out van een query. De twee vullen elkaar aan, dus houd ze allebei.

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

Voor diepere richtlijnen over elk probleem in dit voorbeeld, zie Microsoft Entra authenticatie, Connection pooling, Encryption and certificates, Retry logic en Error handling.

Belangrijkste kenmerken

• PEP 249-conformiteit: standaard connect, cursor, execute, en fetch* interfaces, plus Pythonic-extensies.
• Direct Database Connectivity (DDBC): Geen externe drivermanager vereist. Installeer mssql-python het en je bent klaar om te verbinden.
• Microsoft Entra ID-authenticatie: Ingebouwde ondersteuning voor authenticatiemodi, inclusief beheerde identiteiten en servicehoofden.
• SQL Server en Windows authentication: SQL-logins, Kerberos en Windows single sign-on (SSO) op ondersteunde platforms.
• Bulkkopiëren: Bulkinvoeging met hoge prestaties voor grote gegevensladingen met ondersteuning voor het native TDS-protocol.
• Ondersteuning voor native datatypes: JSON, XML, ruimtelijk, spaarzame kolommen, datetimeoffset en decimaal/money met precieze afhandeling.
• Apache Arrow-integratie: Zero-copy resultaatsets voor snelle gegevensuitwisseling met pandas, Polars en DuckDB.
• Async-patronen: Gebruik de driver met asyncio-gebaseerde applicaties en FastAPI via ThreadPoolExecutor-workarounds. Zie Async-patronen voor integratiepatronen.
• TLS standaard: TLS-encryptie en certificaatvalidatie standaard aan (via ODBC Driver 18). TDS 8.0-versleuteling is beschikbaar wanneer je Encrypt=strict instelt.

Aan de slag

Configureren en authenticeren

Werken met gegevens

SQL Server-gegevenstypen en -functies

Integreer met Python-tools en frameworks

Implementeren en gebruiken

Migreren naar mssql-python

Referentie

Verwante onderwerpen

• mssql-python op GitHub
• mssql-python on PyPI
• Opmerkingen bij de release
• Roadmap
• Azure Python Ontwikkelaarscentrum
• Python Database API Specificatie 2.0