Microsoft Python Driver para SQL Server – mssql-python

mssql-pythoné o driver Python da Microsoft para SQL Server, Banco de Dados SQL do Azure, Instância Gerenciada de SQL do Azure e banco de dados SQL no Microsoft Fabric. Ele usa Conectividade Direta de Banco de Dados (DDBC), então você pode se conectar sem instalar um gerenciador externo de drivers. O driver suporta Python 3.10 ou posterior e está em conformidade com a Especificação 2.0 da API de Banco de Dados Python, além de adicionar melhorias amigáveis para Python no desenvolvimento diário.

Escolha o ponto de partida

• Para fazer um exemplo local de SQL Server rodar rapidamente, comece pelo Quickstart: Conecte com o driver mssql-python.
• Para conectar ao SQL do Azure com autenticação sem senha, comece com a autenticação Microsoft Entra e as strings de conexão.
• Para explorar dados interativamente, comece com Conectar a partir de um notebook Jupyter ou prototipagem rápida.
• Para mover grandes volumes de dados de forma eficiente, vá para Operações de cópia em massa ou o início rápido de cópia em massa.
• Para migrar de outro driver, vá para Migrar de pyodbc, Migrar de pymssql, Migrar de SQLite ou Migrar de PostgreSQL.

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

Use este exemplo como ponto de partida para uma conexão SQL do Azure orientada à produção. Ele lê configurações do ambiente, autentica com identidade gerenciada e habilita a criptografia Tabular Data Stream (TDS) 8.0. Também define tempos limite de login e de consulta por instrução, faz novas tentativas após falhas transitórias com backoff exponencial (uma nova conexão para erros de conexão, a mesma conexão para erros de consulta, como deadlocks), registra os resultados e depende de gerenciadores de contexto para liberar recursos.

As palavras-chave ConnectRetryCount e ConnectRetryInterval na cadeia de conexão habilitam a resiliência a conexões ociosas do SQL Server: o driver reconecta automaticamente, de forma transparente, uma conexão ociosa interrompida. Isso é diferente da nova tentativa no nível da aplicação neste exemplo, que faz uma nova tentativa de uma consulta que falha com um erro transitório, como um deadlock ou tempo limite da consulta. Os dois são complementares, então mantenha os dois.

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 detalhadas sobre cada preocupação neste exemplo, veja Autenticação Microsoft Entra, Pool de conexões, Criptografia e certificados, Lógica de tentativas e Tratamento de erros.

Características principais

• Conformidade com a PEP 249: Interfaces padrão connect, cursor, execute e fetch*, além de extensões idiomáticas do Python.
• Conectividade Direta de Banco de Dados (DDBC): Não é necessário gerenciador externo de drivers. Instale mssql-python e você estará pronto para conectar.
• Autenticação Microsoft Entra ID: Suporte embutido para modos de autenticação, incluindo identidades gerenciadas e princípios de serviço.
• Autenticação do SQL Server e do Windows: logins do SQL Server, Kerberos e logon único do Windows (SSO) em plataformas compatíveis.
• Cópia em massa: Inserção em massa de alto desempenho para grandes cargas de dados com suporte nativo ao protocolo TDS.
• Suporte nativo a tipos de dados: JSON, XML, espacial, colunas esparsas, datetimeoffset e decimal/money com tratamento 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: Use o driver em aplicativos baseados em asyncio e com FastAPI, com soluções alternativas usando o ThreadPoolExecutor. Veja padrões assíncronos para padrões de integração.
• TLS por padrão: criptografia TLS e validação de certificados ativada por padrão (via ODBC Driver 18). A criptografia TDS 8.0 está disponível quando você define Encrypt=strict.

Introdução

Configurar e autenticar

Trabalhar com dados

Tipos e recursos de dados do SQL Server

Integre com ferramentas e frameworks do Python

Implantar e operar

Migrar para mssql-python

Demonstrativo

Conteúdo relacionado

• mssql-python no GitHub
• mssql-python no PyPI
• Notas de lançamento
• Roteiro
• Central de Desenvolvedores do Python do Azure
• Especificação 2.0 da API de Banco de Dados Python