Controlador de Python de Microsoft para SQL Server: mssql-python

mssql-pythones el controlador de Python de Microsoft para SQL Server, Azure SQL Database, Azure SQL Managed Instance y base de datos SQL en Microsoft Fabric. Utiliza Conectividad Directa de Base de Datos (DDBC), por lo que puedes conectarte sin instalar un gestor de controladores externo. El controlador es compatible con Python 3.10 o posterior y cumple con la Especificación 2.0 de la API de Bases de Datos de Python, además de añadir mejoras compatibles con Python para el desarrollo diario.

Elija el punto de partida.

• Para conectarte a SQL Server, Azure SQL Database o SQL Database en Microsoft Fabric y ejecutar tu primera consulta, empieza con Quickstart: Connect with the driver mssql-python.
• Para crear un contenedor local de SQL Server, empieza con el contenedor y el desarrollo local.
• Para conectarte a Azure SQL con autenticación sin contraseña, empieza con la autenticación Microsoft Entra y las cadenas de conexión.
• Para explorar datos de forma interactiva, empieza con Connect from a Jupyter Notebook o prototipado rápido.
• Para mover grandes volúmenes de datos de forma eficaz, vaya a Operaciones de copia masiva o al inicio rápido de copia masiva.
• Para migrar desde otro controlador, ve a Migrar desde pyodbc, Migrar desde pymssql, Migrar desde SQLite o Migrar desde PostgreSQL.

Línea base de producción para Azure SQL

Este ejemplo se conecta a Azure SQL mediante identidad gestionada y cifrado Tabular Data Stream (TDS) 8.0. Lee la configuración de conexión de las variables del entorno y establece tiempos de espera de inicio de sesión y consulta separados. Cada nuevo intento vuelve a llamar a connect() en lugar de reutilizar un objeto de conexión fallida. Los gestores de contexto liberan la conexión y el cursor tras cada intento.

Usa este asistente solo para consultas de lectura idempotentes que sean seguras de repetir. No lo uses para volver a ejecutar escrituras individuales o transacciones parcialmente completadas.

Las palabras clave ConnectRetryCount y ConnectRetryInterval de la cadena de conexión habilitan la resiliencia de las conexiones inactivas en SQL Server: el controlador vuelve a conectar de forma transparente una conexión inactiva interrumpida. Esto es distinto del reintento a nivel de aplicación en esta muestra, que vuelve a intentar una consulta que falla con un error transitorio como un bloqueo o un tiempo de espera de consulta. Los dos son complementarios, así que quédate con ambos.

La comprobación de errores de Azure coincide con el mensaje estándar de error inglés 40501 incluso cuando el diagnóstico omite el número de error. También comprueba ddbc_error si hay números conocidos cuando están presentes. La coincidencia se realiza en la medida de lo posible: los mensajes localizados o con formato diferente pueden no coincidir. Para obtener más información, consulte errores transitorios de conexión.

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 una orientación más profunda sobre cada preocupación en este ejemplo, véase autenticación Microsoft Entra, Agrupación de conexiones, Cifrado y certificados, Lógica de reintentos y Manejo de errores.

Características clave

• Cumplimiento de PEP 249: interfaces connect, cursor, execute y fetch* estándar, además de extensiones propias de Python.
• Conectividad directa con bases de datos (DDBC): No se requiere gestor externo de controladores. Instala mssql-python y estarás listo para conectar.
• Autenticación de Microsoft Entra ID: Compatibilidad integrada con modos de autenticación, incluidos identidades administradas y entidades de servicio.
• SQL Server y autenticación de Windows: inicios de sesión de SQL, Kerberos y inicio de sesión único (SSO) de Windows en plataformas compatibles.
• Copia masiva: Inserción masiva de alto rendimiento para grandes cargas de datos con soporte nativo del protocolo TDS.
• Soporte nativo de tipos de datos: JSON, XML, espacial, columnas dispersas, desfase de fecha y hora y decimal/dinero con manejo preciso.
• Integración con Apache Arrow: Conjuntos de resultados sin copia para un intercambio rápido de datos con pandas, Polar y DuckDB.
• Patrones asíncronos: Usa el driver con aplicaciones basadas en asyncio y FastAPI mediante soluciones alternativas con ThreadPoolExecutor. Consulta patrones asíncronos para patrones de integración.
• TLS por defecto: cifrado TLS y validación de certificados activados por defecto con cualquier proveedor ODBC nativo. El cifrado TDS 8.0 está disponible cuando configuras Encrypt=strict.

Comienza

Configurar y autenticar

Trabajar con datos

Tipos de datos y características de SQL Server

Integra con herramientas y frameworks de Python

Despliegue y operación

Migrar a mssql-python

Solicitud de una característica

Para solicitar una funcionalidad, abre un problema en el repositorio de GitHub de mssql-python.

Contenido relacionado

• Ciclo de vida de soporte para mssql-python
• Novedades en mssql-python
• mssql-python en GitHub
• mssql-python en PyPI
• Notas de lanzamiento
• Plan de desarrollo
• Centro para desarrolladores de Python para Azure
• Especificación 2.0 de la API de Bases de Datos de Python