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 far partire rapidamente un esempio locale di SQL Server, inizia con Quickstart: Collegati con il driver mssql-python.
• 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

Usa questo esempio come punto di partenza per una connessione Azure SQL orientata alla produzione. Legge la configurazione dall'ambiente, si autentica con l'identità gestita e abilita la crittografia Tabular Data Stream (TDS) 8.0. Imposta inoltre timeout di accesso e timeout delle query per istruzione, ritenta i fallimenti transitori con un backoff esponenziale (una nuova connessione per gli errori di connessione, la stessa connessione per errori di query come i deadlock), registra gli esiti e si affida ai context manager per rilasciare le risorse.

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.

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

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 (tramite ODBC Driver 18). 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

Riferimento

Contenuti correlati

• 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