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 que una muestra local de SQL Server funcione rápidamente, empieza con Inicio rápido: Conéctate con el controlador mssql-python.
• 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

Utiliza este ejemplo como punto de partida para una conexión Azure SQL orientada a producción. Lee la configuración del entorno, se autentica con la identidad gestionada y habilita el cifrado Tabular Data Stream (TDS) 8.0. También establece tiempos de espera de inicio de sesión y de consulta por sentencia, vuelve a intentar fallos transitorios con retroceso exponencial (una conexión nueva para errores de conexión, la misma conexión para errores de consulta como bloqueos), registra los resultados y depende de gestores de contexto para liberar recursos.

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.

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 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: el cifrado TLS y la validación de certificados activados por defecto (a través del controlador ODBC 18). 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

Referencia

Contenido relacionado

• 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