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 se ligar ao SQL Server, Base de Dados SQL do Azure ou SQL Database no Microsoft Fabric e executar a sua primeira consulta, comece pelo Quickstart: Ligue-se ao driver mssql-python.
• Para criar um contentor SQL Server local, comece pelo contentor e desenvolvimento local.
• 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

Este exemplo liga-se ao SQL do Azure com identidade gerida e encriptação Tabular Data Stream (TDS) 8.0. Lê as definições de ligação a partir de variáveis de ambiente e define tempos de login e consulta separados. Cada nova tentativa chama connect() novamente em vez de reutilizar um objeto de ligação falhado. Os context managers libertam a ligação e o cursor após cada tentativa.

Use este assistente apenas para consultas de leitura idempotentes que sejam seguras para repetir. Não o uses para reexecutar operações de escrita individuais ou transações parcialmente concluídas.

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.

A verificação de erro do Azure corresponde à mensagem padrão em inglês 40501 mesmo quando o diagnóstico omite o número de erro. Também verifica se há números conhecidos em ddbc_error quando presentes. A correspondência é o melhor esforço: mensagens localizadas ou com formato diferente podem não corresponder. Para mais informações, consulte Erros de ligação transitórios.

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

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 ativadas por defeito com qualquer um dos dois fornecedores ODBC nativos. 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

Pedir uma funcionalidade

Para solicitar uma funcionalidade, abra um problema no repositório GitHub mssql-python.

Conteúdo relacionado

• Ciclo de vida de suporte para mssql-python
• Novidades no mssql-python
• 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