Microsoft Python Driver pour SQL Server - mssql-python

mssql-pythonest le pilote Python de Microsoft pour SQL Server, Azure SQL Database, Azure SQL Managed Instance et la base de données SQL dans Microsoft Fabric. Il utilise la connectivité directe à base de données (DDBC), ce qui permet de vous connecter sans installer de gestionnaire de pilotes externe. Le pilote prend en charge Python 3.10 ou versions ultérieures et est conforme à la spécification API de base de données Python 2.0 tout en ajoutant des améliorations adaptées à Python pour le développement quotidien.

Choisir votre point de départ

• Pour faire tourner rapidement un exemple local de SQL Server, commencez par Quickstart : Connectez-vous au pilote mssql-python.
• Pour se connecter à Azure SQL avec une authentification sans mot de passe, commencez par l’authentification Microsoft Entra et les chaînes de connexion.
• Pour explorer les données de manière interactive, commencez par Connect from a Jupyter Notebook ou le prototypage rapide.
• Pour déplacer efficacement de grands volumes de données, passez à Opérations de copie en masse ou au démarrage rapide de la copie en masse.
• Pour migrer depuis un autre pilote, allez dans Migrer depuis pyodbc, migrer depuis pymssql, migrer depuis SQLite, ou migrer depuis PostgreSQL.

Base de référence de production pour Azure SQL

Utilisez cet exemple comme point de départ pour une connexion Azure SQL orientée production. Il lit la configuration de l’environnement, s’authentifie avec une identité gérée, et permet le chiffrement Tabular Data Stream (TDS) 8.0. Il définit également les délais de connexion et de requête par état, réessaie les échecs transitoires avec un retour exponentiel (une connexion fraîche pour les erreurs de connexion, la même connexion pour les erreurs de requête comme les blocages), enregistre les résultats et s’appuie sur les gestionnaires de contexte pour libérer les ressources.

Les mots-clés ConnectRetryCount et ConnectRetryInterval de la chaîne de connexion activent la résilience des connexions inactives de SQL Server : le pilote rétablit automatiquement une connexion inactive interrompue. Cela se distingue de la réévaluation au niveau de l’application dans cet exemple, qui retente une requête échouant avec une erreur transitoire telle qu’un blocage ou un délai d’attente de requête. Les deux sont complémentaires, donc gardez les deux.

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

Pour des conseils plus approfondis sur chaque problème de cet exemple, voir authentification Microsoft Entra, Pooling de connexions, Chiffrement et certificats, Logique de tentative et Gestion des erreurs.

Principales fonctionnalités

• Conformité à la PEP 249 : interfaces standard connect, cursor, execute et fetch*, ainsi que des extensions pythoniques.
• Connectivité directe à la base de données (DDBC) : Aucun gestionnaire de pilotes externe requis. Installez mssql-python et vous êtes prêt à vous connecter.
• Authentification Microsoft Entra ID : prise en charge intégrée des modes d’authentification, y compris les identités managées et les principaux de service.
• SQL Server et Authentification Windows : connexions SQL, Kerberos et connexion unique (SSO) Windows sur les plateformes prises en charge.
• Copie en bloc : insertion en masse à hautes performances pour de gros volumes de données avec prise en charge native du protocole TDS.
• Prise en charge des types de données natifs : JSON, XML, spatial, colonnes éparses, datetimeoffset et decimal/money avec une gestion précise.
• Intégration Apache Arrow : ensembles de résultats zéro copie pour un échange rapide de données avec pandas, Polar et DuckDB.
• Modèles asynchrones : utilisez le pilote avec des applications basées sur asyncio et FastAPI, via des solutions de contournement avec ThreadPoolExecutor. Voir Motifs asynchrones pour les motifs d’intégration.
• TLS par défaut : le chiffrement TLS et la validation des certificats sont activés par défaut (via le pilote ODBC 18). Le chiffrement TDS 8.0 est disponible lorsque vous définissez Encrypt=strict.

Commencez

Configurer et authentifier

Utiliser des données

Types de données et fonctionnalités de SQL Server

Intégrer avec des outils et frameworks Python

Déployer et exploiter

Migrer vers mssql-python

Référence

Contenu connexe

• mssql-python sur GitHub
• mssql-python sur PyPI
• Notes de publication
• Feuille de route
• Centre de développement Azure Python
• Spécification 2.0 de l’API de base de données Python