Driver do Microsoft Python para SQL Server - mssql-python

mssql-pythoné o driver Python da Microsoft para SQL Server, Base de Dados SQL do Azure, Azure SQL Managed Instance e base de dados SQL no Microsoft Fabric. Utiliza Direct Database Connectivity (DDBC), por isso pode ligar-se sem instalar um gestor de drivers externo. O driver suporta Python 3.10 ou posterior e cumpre a Especificação 2.0 da API de Bases de Dados de Python, adicionando melhorias amigáveis para Python no desenvolvimento diário.

Escolhe o teu ponto de partida

• Para fazer um exemplo local do SQL Server correr rapidamente, comece pelo Quickstart: Ligue-se ao driver mssql-python.
• Para se ligar ao SQL do Azure com autenticação sem palavra-passe, comece com autenticação Microsoft Entra e strings de conexão.
• Para explorar dados de forma interativa, comece com Ligar a partir de um Jupyter Notebook ou prototipagem rápida.
• Para mover grandes volumes de dados de forma eficiente, vá para operações de cópia em massa ou para o início rápido de cópia em massa.
• Para migrar de outro driver, vá a Migrar de pyodbc, Migrar de pymssql, Migrar de SQLite ou Migrar de PostgreSQL.

Linha de base de produção para o SQL do Azure

Use este exemplo como ponto de partida para uma ligação SQL do Azure orientada à produção. Lê a configuração do ambiente, autentica com identidade gerida e permite a encriptação Tabular Data Stream (TDS) 8.0. Também define tempos limite de início de sessão e de consulta por instrução, volta a tentar após falhas transitórias com recuo exponencial (uma nova ligação para erros de ligação, a mesma ligação para erros de consulta, como deadlocks), regista os resultados e recorre a gestores de contexto para libertar recursos.

As palavras-chave ConnectRetryCount e ConnectRetryInterval na cadeia de ligação permitem a resiliência de ligações inativas do SQL Server: o controlador volta a ligar de forma transparente uma ligação inativa interrompida. Isto é diferente da nova tentativa ao nível da aplicação neste exemplo, que repete uma consulta que falha com um erro transitório, como um deadlock ou o tempo limite da consulta. Os dois são complementares, por isso mantém ambos.

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

Para orientações mais profundas sobre cada preocupação neste exemplo, consulte autenticação Microsoft Entra, Pool de ligações, Encriptação e certificados, Lógica de Retentativas e Tratamento de erros.

Principais características

• Conformidade com a PEP 249: Interfaces connect, cursor, execute e fetch* padrão, bem como extensões pitónicas.
• Conectividade Direta à Base de Dados (DDBC): Não é necessário gestor externo de drivers. Instala mssql-python e estás pronto para ligar.
• Autenticação Microsoft Entra ID: Suporte integrado para modos de autenticação, incluindo identidades geridas e princípios de serviço.
• SQL Server e autenticação do Windows: inícios de sessão SQL, Kerberos e single sign-on (SSO) do Windows em plataformas suportadas.
• Cópia em massa: Inserção em massa de alto desempenho para grandes cargas de dados com suporte nativo do protocolo TDS.
• Suporte nativo para tipos de dados: JSON, XML, espaciais, colunas esparsas, datetimeoffset e decimal/moeda com processamento preciso.
• Integração com Apache Arrow: Conjuntos de resultados sem cópias para troca rápida de dados com pandas, Polars e DuckDB.
• Padrões assíncronos: Utilize o driver com aplicações baseadas em asyncio e o FastAPI, recorrendo a soluções alternativas com o ThreadPoolExecutor. Veja padrões assíncronos para padrões de integração.
• TLS por defeito: encriptação TLS e validação de certificados ativada por defeito (via ODBC Driver 18). A encriptação TDS 8.0 está disponível quando defines Encrypt=strict.

Introdução

Configurar e autenticar

Trabalhar com dados

Tipos de dados e funcionalidades do SQL Server

Integrar com ferramentas e frameworks de Python

Implantar e operar

Migrar para mssql-python

Referência

Conteúdo relacionado

• mssql-python no GitHub
• mssql-python no PyPI
• Notas de lançamento
• Plano de Ação
• Centro de Desenvolvedores do Azure Python
• Especificação 2.0 da API de Base de Dados Python