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.

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 instrucción, 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;"
        # 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:
        # 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

Artículo Descripción
Instalación Instala mssql-python y verifica tu entorno Python.
Inicio rápido: Conéctate con mssql-python Conéctate a una instancia local o prueba SQL Server y ejecuta tu primera consulta.
Inicio rápido: Conectar desde un Jupyter Notebook Usa mssql-python dentro de un cuaderno para una exploración interactiva de datos.
Inicio rápido: Copia masiva Mueve grandes conjuntos de datos a SQL Server con la API de copia masiva.
Inicio rápido: Prototipado rápido Crea scripts pequeños y pruebas de concepto rápidamente.
Inicio rápido: Despliegues repetibles Empaqueta, configura y lanza aplicaciones Python que se comuniquen con SQL.
Inicio rápido de Apache Arrow Obtener resultados de consultas como tablas de flechas Apache para flujos de trabajo analíticos.

Configurar y autenticar

Artículo Descripción
Cadenas de conexión Sintaxis de cadenas de conexión, palabras clave comunes y ejemplos.
Construir cadenas de conexión de forma programática Crea cadenas de conexión de forma segura a partir de la configuración y los secretos.
Administración de conexiones Abrir, reutilizar y cerrar las conexiones de forma limpia.
Agrupación de conexiones Optimización del pool, ciclos de vida y patrones de reutilización.
Cifrado y certificados Modos de cifrado TLS, validación de certificados y TDS 8.0.
Autenticación de Microsoft Entra Autenticación sin contraseña para Azure SQL con identidad gestionada, principal de servicio, flujos interactivos y código de dispositivo.
Recomendaciones de seguridad Parametrización, administración de secretos, privilegios mínimos y cifrado.
Grupos de disponibilidad Conéctese a los grupos de disponibilidad Always On y a las réplicas de solo lectura.

Trabajar con datos

Artículo Descripción
Ejecución de consultas execute, executemany, lotes de varias sentencias y conjuntos de resultados.
Recuperación de datos fetchone, fetchmany, fetchall, y patrones de streaming.
Consultas con parámetros Vincula parámetros de forma segura para evitar la inyección SQL.
procedimientos almacenados Llamar a procedimientos, leer parámetros de salida y procesar conjuntos de resultados.
Gestión del cursor Duración de la vida del cursor, desplazamiento y ajuste del tamaño de la matriz.
Objetos de fila Acceda a las filas por índice, nombre o como asignaciones.
Administración de transacciones Comprometerse, retroceder, guardar puntos y niveles de aislamiento.
Paginación Patrones de paginación por conjuntos clave y por desplazamiento para grandes conjuntos de resultados.
Control de errores mssql_python.Error, DatabaseError, y la estructura de error de SQL Server.
Lógica de reintento Detectar errores transitorios y reintentar con retroceso exponencial.

Tipos de datos y características de SQL Server

Artículo Descripción
Asignaciones de tipos de datos Tablas y reglas de conversión tipo SQL Server a Python.
Gestión de fecha y hora datetime, datetime2, datetimeoffset, y consideraciones de huso horario.
Decimal y tipos de moneda Tipos numéricos exactos y precisión decimal.Decimal.
Cadenas y datos Unicode varchar, nvarchar, intercalaciones y páginas de códigos.
Gestión de NULL Lógica de tres valores, centinelas y pandas interoperan.
Datos binarios varbinary, image y transmisión en flujo de objetos grandes.
Convertidores de tipo personalizado Registrar convertidores de entrada y salida para tipos personalizados.
Operaciones de copia masiva Inserciones de alto rendimiento con la API de copia en bloque.
Datos JSON Almacena, consulta y triturar JSON con FOR JSON y OPENJSON.
Datos XML Trabaja con el xml tipo de dato, XPath y XQuery.
Datos espaciales geometry y tipos geography de Python.
Columnas dispersas Columnas dispersas y conjuntos de columnas para tablas anchas.
Detección de esquemas Inspecciona bases de datos, tablas, columnas e índices.

Integra con herramientas y frameworks de Python

Artículo Descripción
Integración con Apache Arrow Obtener resultados como tablas Arrow para análisis de copia cero.
Integración con pandas Carga los resultados de la consulta en DataFrames y escríbelos de nuevo.
Integración con Polars Usa Polars con mssql-python para cargas de trabajo por columnas.
Integración con DuckDB Consulta los datos de SQL Server junto con tablas locales de DuckDB.
Integración con FastAPI Conecta mssql-python a los servicios FastAPI.
Integración con Flask Usa mssql-python en aplicaciones Flask.
Patrones asincrónicos Combina mssql-python con asyncio y grupos de subprocesos.
Acceso a datos y patrones de análisis Elige la ruta de lectura correcta para el acceso mediante cursor, la extracción con Arrow, pandas, Polars y el análisis con DuckDB sobre datos SQL.
Carga de datos y patrones de movimiento Elija la ruta de escritura adecuada para inserciones de filas, copia en bloque, operaciones de inserción o actualización MERGE, carga de DataFrame e ingesta CSV.

Despliegue y operación

Artículo Descripción
Contenedor y desarrollo local Configura contenedores Docker, devcontainers y pipelines de CI para aplicaciones Python que se conecten a SQL.
Ajuste del rendimiento Ajuste de grupos, instrucciones preparadas, tamaños de lote y copia en bloque.
Solución de problemas Errores comunes, registro y diagnósticos de certificados.
Configuración del módulo Configuración de nivel de módulo, hooks de registro y banderas de funcionalidad.

Migrar a mssql-python

Artículo Descripción
Migrar desde pyodbc Asignar las API pyodbc y cadenas de conexión a mssql-python.
Migrar de pymssql Sustituye pymssql por mssql-python manteniendo el comportamiento.
Migración desde SQLite Mueve cargas de trabajo locales de SQLite a SQL Server o Azure SQL.
Migración desde PostgreSQL Guía integral para desarrolladores de Python que pasan de PostgreSQL a SQL Server con mssql-python.

Contenido relacionado

  • Last updated on