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 verbinding te maken met SQL Server, Azure SQL Database of SQL-database in Microsoft Fabric en je eerste query uit te voeren, start je met Quickstart: Maak verbinding met de mssql-python-driver.
• Om een lokale SQL Server-container te maken, begin je met Container en lokale ontwikkeling.
• 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

Dit voorbeeld maakt verbinding met Azure SQL met beheerde identiteit en Tabular Data Stream (TDS) 8.0-encryptie. Het leest verbindingsinstellingen uit omgevingsvariabelen en stelt aparte inlog- en query-timeouts in. Elke herpoging roept connect() opnieuw aan in plaats van een mislukt verbindingsobject opnieuw te gebruiken. contextmanagers geven de verbinding en cursor na elke poging vrij.

Gebruik deze helper alleen voor idempotente leeszoekopdrachten die veilig herhaald kunnen worden. Gebruik het niet om individuele schrijfopdrachten of gedeeltelijk voltooide transacties opnieuw af te spelen.

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.

De Azure-foutcontrole komt overeen met het standaard Engelse foutbericht 40501, zelfs wanneer de diagnose het foutnummer weglaat. Het controleert ddbc_error ook op bekende getallen wanneer aanwezig. Matching is best effort: gelokaliseerde of anders opgemaakte berichten komen mogelijk niet overeen. Zie Tijdelijke verbindingsfouten voor meer informatie.

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

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 ingeschakeld: TLS-encryptie en certificaatvalidatie staan standaard aan bij een van beide native ODBC-providers. 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

Verzoeken om een functie toe te voegen

Om een functie aan te vragen, open je een issue in de mssql-python GitHub-repository.

Verwante onderwerpen

• Ondersteuningslevenscyclus voor mssql-python
• Wat is er nieuw in mssql-python
• mssql-python op GitHub
• mssql-python on PyPI
• Opmerkingen bij de release
• Roadmap
• Azure Python Ontwikkelaarscentrum
• Python Database API Specificatie 2.0