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

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

Este exemplo se conecta ao SQL do Azure com identidade gerenciada e criptografia Tabular Data Stream (TDS) 8.0. Ele lê as configurações de conexão das variáveis do ambiente e define tempos separados de login e consulta. Cada nova tentativa chama connect() novamente em vez de reutilizar um objeto de conexão com falha. Gerenciadores de contexto liberam a conexão e o cursor após cada tentativa.

Use este auxiliar apenas para consultas de leitura idempotentes que podem ser repetidas com segurança. Não use para reexecutar operações de escrita individuais ou transações parcialmente concluídas.

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.

A verificação de erro do Azure corresponde à mensagem padrão de erro em inglês 40501 mesmo quando o diagnóstico omite o número do erro. Também verifica ddbc_error em busca de números conhecidos, quando presentes. A correspondência é o melhor esforço: mensagens localizadas ou formatadas de forma diferente podem não corresponder. Para mais informações, veja Erros de conexão transitória.

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

Introdução

Artigo Descrição
Instalação Instale mssql-python e verifique seu ambiente Python.
Início rápido: Conecte-se com mssql-python Conecte-se a uma instância local ou teste do SQL Server e execute sua primeira consulta.
Início Rápido: Conecte-se a partir de um Jupyter Notebook Use mssql-python dentro de um caderno para exploração interativa de dados.
Início rápido: Cópia em massa Mova grandes conjuntos de dados para o SQL Server com a API de cópia em massa.
Início Rápido: Prototipagem rápida Construa scripts pequenos e provas de conceito rapidamente.
Início Rápido: Implantações repetíveis Empacote, configure e envie aplicações Python que se comunicam com SQL.
Início rápido do Apache Arrow Buscar resultados de consulta como tabelas Apache Arrow para fluxos de trabalho analíticos.

Configurar e autenticar

Artigo Descrição
Provedor mssql-odbc (Prévia) Selecione o provedor nativo de pré-visualização e revise os recursos suportados e problemas conhecidos.
Strings de conexão Sintaxe de string de conexão, palavras-chave comuns e exemplos.
Construir cadeias de conexão programáticamente Monte strings de conexão com segurança com base em configurações e segredos.
Gerenciamento de conexões Abra, reutilize e feche as conexões de forma limpa.
Agrupamento de conexões Ajuste de pool, tempos de vida e padrões de reutilização.
Criptografia e certificados Modos de criptografia TLS, validação de certificados e TDS 8.0.
Autenticação do Microsoft Entra Autenticação sem senha para o SQL do Azure com identidade gerenciada, entidade de serviço, fluxos interativos e fluxos de código do dispositivo.
Melhores práticas de segurança Parametrização, gerenciamento de segredos, privilégio mínimo e criptografia.
Grupos de disponibilidade Conecte-se aos grupos de disponibilidade Always On e às réplicas somente de leitura.

Trabalhar com dados

Artigo Descrição
Execução de consultas execute, executemany, lotes com múltiplas instruções e conjuntos de resultados.
Recuperação de dados fetchone, fetchmany, fetchall, e padrões de streaming.
Consultas parametrizadas Vincule parâmetros com segurança para evitar a injeção de SQL.
Procedimentos armazenados Chame procedimentos, leia os parâmetros de saída e processe conjuntos de resultados.
Gerenciamento do cursor Tempos de vida do cursor, rolagem e ajuste do arraysize.
Objetos de linha Acesse as linhas por índice, nome ou como mapeamentos.
Gerenciamento de transações Commit, reversão, pontos de salvamento e níveis de isolamento.
Paginação Padrões de paginação por keyset e por deslocamento em grandes conjuntos de resultados.
Tratamento de erros mssql_python.Error, DatabaseError, e estrutura de erro do SQL Server.
Lógica de repetição Detecte erros transitórios e tente novamente com retardo exponencial.

Tipos e recursos de dados do SQL Server

Artigo Descrição
Mapeamentos de tipo de dados Tabelas e regras de conversão do tipo SQL Server para Python.
Manipulação de datas datetime, datetime2, datetimeoffset, e considerações de fuso horário.
Tipos decimais e moeda Tipos numéricos exatos e precisão decimal.Decimal.
Cadeia de caracteres e dados Unicode varchar, nvarchar, colações e páginas de código.
Tratamento de NULL Lógica de três valores, sentinelas e pandas interoperam.
Dados binários varbinary, image e transmissão de objetos grandes.
Conversores de tipo personalizado Registrem conversores de entrada e saída para tipos personalizados.
Operações de cópia em massa Inserções de alta taxa de transferência com a API de cópia em lote.
Dados JSON Armazene, consulte e fragmente JSON com FOR JSON e OPENJSON.
Dados XML Trabalhe com o xml tipo de dado, XPath e XQuery.
Dados espaciais geometry e os tipos geography do Python.
Colunas esparsas Colunas esparsas e conjuntos de colunas para tabelas amplas.
Descoberta de esquema Inspecione bancos de dados, tabelas, colunas e índices.

Integre com ferramentas e frameworks do Python

Artigo Descrição
Integração com Apache Arrow Busque resultados como tabelas Arrow para análises sem cópia.
integração do Pandas Carregue os resultados das consultas em DataFrames e grave-os novamente.
Integração com Polars Use Polars com mssql-python para cargas de trabalho em colunas.
Integração com DuckDB Consulte dados do SQL Server junto com tabelas locais do DuckDB.
Integração com FastAPI Conecte mssql-python aos serviços FastAPI.
Teste e implante aplicações FastAPI Configure agrupamento, erros, autenticação, testes e configurações de implantação para serviços do FastAPI.
Integração com Flask Use mssql-python em aplicações Flask.
Padrões assíncronos Combine o mssql-python com asyncio e conjuntos de threads.
Padrões de acesso a dados e análises Escolha o caminho de leitura correto para acesso ao cursor, extração de setas, pandas, Polars e análises do DuckDB sobre dados do SQL.
Padrões de carregamento e movimento de dados Escolha o caminho de escrita correto para inserções de linhas, cópia em massa, MERGE upserts, carregamento de DataFrame e ingestão de CSV.

Implantar e operar

Artigo Descrição
Contêiner e desenvolvimento local Configure contêineres do Docker, devcontainers e pipelines de CI para aplicativos do Python que se conectam ao SQL.
Otimização do desempenho Ajuste de pool, instruções preparadas, tamanhos de lote e cópia em massa.
Solução de problemas Encontre orientações por sintoma ou mensagem de erro.
Instalação e solução de problemas de conexão Resolver problemas de instalação, conexão, container e CI.
Consulta de dados, solução de problemas de operações Resolver questões de consulta, tipo de dado, desempenho, transação e cópia em massa.
Configuração do módulo Configurações em nível de módulo, ganchos de registro e sinalizadores de recurso.

Migrar para mssql-python

Artigo Descrição
Migrar do pyodbc Mapeie as APIs do pyodbc e as strings de conexão para o mssql-python.
Migrar de pymssql Substitua pymssql por mssql-python preservando o comportamento.
Migrar a partir do SQLite Mova cargas de trabalho locais do SQLite para o SQL Server ou SQL do Azure.
Migrar do PostgreSQL Guia único para desenvolvedores Python que migram do PostgreSQL para o SQL Server com mssql-python.

Solicitar um recurso

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

Conteúdo relacionado

  • Última atualização em