Driver Microsoft Python per SQL Server - mssql-python

mssql-pythonè il driver Python di Microsoft per SQL Server, database SQL di Azure, Istanza gestita di SQL di Azure e database SQL in Microsoft Fabric. Utilizza la connettività diretta al database (DDBC), quindi puoi connetterti senza installare un gestore driver esterno. Il driver supporta Python 3.10 o successivo e rispetta la specifica API di database Python 2.0, aggiungendo miglioramenti compatibili con Python per lo sviluppo quotidiano.

Scegliere il punto di partenza

• Per connetterti a SQL Server, database SQL di Azure o SQL database in Microsoft Fabric ed eseguire la tua prima query, inizia con Quickstart: Connettiti con il driver mssql-python.
• Per creare un container SQL Server locale, inizia con Container e sviluppo locale.
• Per connetterti ad Azure SQL con autenticazione senza password, inizia con l'autenticazione Microsoft Entra e le stringhe di connessione.
• Per esplorare i dati in modo interattivo, inizia con Connect from a Jupyter Notebook o prototipazione rapida.
• Per spostare grandi volumi di dati in modo efficiente, vai a Operazioni di copia in blocco o alla guida introduttiva alla copia in blocco.
• Per migrare da un altro driver, vai su Migra da pyodbc, Migra da pymssql, Migra da SQLite o Migra da PostgreSQL.

Baseline di produzione per Azure SQL

Questo esempio si collega ad Azure SQL con identità gestita e crittografia Tabular Data Stream (TDS) 8.0. Legge le impostazioni di connessione dalle variabili dell'ambiente e imposta timeout separati per login e query. Ogni ritentativo chiama connect() di nuovo invece di riutilizzare un oggetto di connessione fallito. I gestori di contesto rilasciano la connessione e il cursore dopo ogni tentativo.

Usa questa funzione di supporto solo per query di lettura idempotenti che sono sicure da ripetere. Non usarlo per rieseguire singole operazioni di scrittura o transazioni parzialmente completate.

Le parole chiave ConnectRetryCount e ConnectRetryInterval nella stringa di connessione consentono a SQL Server la resilienza delle connessioni inattive: il driver riconnette in modo trasparente una connessione inattiva interrotta. Questo è diverso dal nuovo tentativo a livello applicativo in questo esempio, che esegue nuovamente una query che non riesce a causa di un errore transitorio, ad esempio un deadlock o un timeout della query. I due sono complementari, quindi tieni entrambi.

Il controllo errore di Azure corrisponde al messaggio standard inglese 40501 anche quando la diagnostica omette il numero di errore. Controlla ddbc_error anche i numeri noti quando presenti. La corrispondenza avviene in modalità best effort: messaggi localizzati o formattati diversamente potrebbero non corrispondere. Per ulteriori informazioni, vedere Errori di connessione temporanei.

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

Per indicazioni più approfondite su ciascuna problematica in questo esempio, vedi Microsoft Entra autenticazione, Connection pooling, Crittografia e certificati, Logica dei ritenti e Gestione degli errori.

Funzionalità principali

• Conformità alla PEP 249: interfacce standard connect, cursor, execute e fetch*, oltre a estensioni pythoniche.
• Connettività diretta al database (DDBC): Nessun gestore esterno di driver richiesto. Installa mssql-python e sei pronto per connetterti.
• Autenticazione di Microsoft Entra ID: Supporto integrato per modalità di autenticazione, tra cui identità gestite e entità servizio.
• SQL Server e autenticazione di Windows: login SQL, Kerberos e single sign-on (SSO) di Windows su piattaforme supportate.
• Copia in massa: inserimento bulk ad alte prestazioni per carichi dati di grandi dimensioni con supporto nativo al protocollo TDS.
• Supporto nativo dei tipi di dati: JSON, XML, spaziale, colonne sparse, datetimeoffset e decimale/money con gestione precisa.
• Integrazione Apache Arrow: set di risultati senza copia per uno scambio rapido di dati con pandas, Polars e DuckDB.
• Pattern asincroni: Utilizza il driver con applicazioni basate su asyncio e FastAPI tramite soluzioni alternative con ThreadPoolExecutor. Vedi Pattern asincroni per i pattern di integrazione.
• TLS di default: crittografia TLS e validazione dei certificati attivati di default con uno dei due fornitori ODBC nativi. La crittografia TDS 8.0 è disponibile quando si imposta Encrypt=strict.

Inizia subito

Configura e autentica

Uso dei dati

Tipi di dati e caratteristiche di SQL Server

Integra con strumenti e framework Python

Distribuire e gestire

Migra a mssql-python

Richiedere una funzionalità

Per richiedere una funzionalità, apri un problema nel repository GitHub di mssql-python.

Contenuti correlati

• Ciclo di vita di supporto per mssql-python
• Novità in mssql-python
• mssql-python su GitHub
• mssql-python su PyPI
• Note di rilascio
• Roadmap
• Centro per sviluppatori Python di Azure
• Specifica API di database Python 2.0