Solucionar problemas de consultas, datos y operación con mssql-python

Utiliza este artículo para diagnosticar problemas de ejecución de consultas, tipo de datos, rendimiento, transacciones y copias masivas con el mssql-python controlador.

Problemas de ejecución de consultas

Tabla u objeto no encontrado

Síntomas:

ProgrammingError: [42S02] (208) Invalid object name 'TableName'.

Posibles causas y soluciones:

  • Contexto incorrecto de la base de datos

    cursor.execute("SELECT DB_NAME()")
print(cursor.fetchone()[0])

  • Esquema no especificado

    cursor.execute("SELECT * FROM dbo.TableName")

  • La tabla no existe

    cursor.execute("""
    SELECT TABLE_NAME
    FROM INFORMATION_SCHEMA.TABLES
    WHERE TABLE_NAME = 'TableName'
""")

Error de sintaxis

Síntomas:

ProgrammingError: [42000] (102) Incorrect syntax near '...'.

Soluciones:

  1. Prueba la sentencia SQL en SQL Server Management Studio (SSMS) para verificar la sintaxis.

  2. Utiliza una consulta parametrizada en lugar de interpolación de cadenas:

    # Don't use string interpolation for query parameters.
cursor.execute(f"SELECT * FROM Production.Product WHERE Name = '{name}'")

# Use parameters.
cursor.execute(
    "SELECT * FROM Production.Product WHERE Name = %(name)s",
    {"name": name},
)

Errores de parámetros

Síntomas:

ProgrammingError: [07001] Wrong number of parameters

Soluciones:

  1. Cuenta los marcadores de posición y los parámetros. Los conteos deben coincidir.

  2. Elige el estilo de parámetro correcto:

    # Qmark style: positional parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = ? AND Name LIKE ?",
    (1, "Adjustable%"),
)
print(cursor.fetchone())

# Pyformat style: named parameters
cursor.execute(
    "SELECT * FROM Production.Product "
    "WHERE ProductID = %(id)s AND Name LIKE %(name)s",
    {"id": 1, "name": "Adjustable%"},
)
print(cursor.fetchone())

Problemas con el tipo de datos

Errores de conversión de fecha y hora

Síntomas:

DataError: [22007] Invalid datetime format

Solution:

Usa objetos en Python datetime en lugar de cadenas.

from datetime import datetime

cursor.execute("CREATE TABLE #Events (EventDate DATETIME)")

# This value raises an error because the date is invalid.
try:
    cursor.execute(
        "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
        {"event_date": "2024-13-45"},
    )
except Exception as e:
    print(f"Expected error: {e}")

# Use a Python datetime object.
cursor.execute(
    "INSERT INTO #Events (EventDate) VALUES (%(event_date)s)",
    {"event_date": datetime(2024, 3, 15)},
)
cursor.execute("SELECT EventDate FROM #Events")
print(cursor.fetchone())

Problemas de precisión decimal

Síntomas:

Los números aparecen truncados o redondeados incorrectamente.

Solution:

Uso decimal.Decimal para valores numéricos precisos:

from decimal import Decimal

cursor.execute("CREATE TABLE #PriceDemo (ListPrice DECIMAL(10, 2))")
cursor.execute(
    "INSERT INTO #PriceDemo (ListPrice) VALUES (%(list_price)s)",
    {"list_price": Decimal("19.99")},
)

Problemas de codificación Unicode

Síntomas:

Los caracteres especiales aparecen distorsionados o causan errores.

Soluciones:

  1. Usa columnas nvarchar para los datos Unicode en tu base de datos.

  2. Pasa las cadenas directamente. El controlador gestiona la codificación:

    cursor.execute("CREATE TABLE #UnicodeDemo (Name NVARCHAR(50))")
cursor.execute(
    "INSERT INTO #UnicodeDemo (Name) VALUES (%(name)s)",
    {"name": "日本語"},
)
cursor.execute("SELECT Name FROM #UnicodeDemo")
print(cursor.fetchone())

Problemas de rendimiento

Ejecución lenta de consultas

Posibles causas y soluciones:

  • Índices faltantes: Revisa el plan de ejecución de consultas en SSMS.

  • Conjuntos de resultados grandes: Usar fetchmany() en lugar de fetchall():

    cursor.arraysize = 1000
while True:
    rows = cursor.fetchmany()
    if not rows:
        break
    process_rows(rows)

  • Agrupación de conexiones desactivada: Habilitar agrupación:

    import mssql_python

mssql_python.pooling(max_size=20, idle_timeout=300)

Problemas de memoria con resultados grandes

Síntomas:

El proceso de Python se queda sin memoria.

Soluciones:

  1. Transmita los resultados en flujo en lugar de cargar todas las filas en la memoria.

    cursor.execute("SELECT * FROM LargeTable")
for row in cursor:
    process_row(row)

  2. Usa la paginación del lado del servidor.

    page_size = 1000
offset = 0

while True:
    cursor.execute(
        "SELECT * FROM LargeTable ORDER BY ID "
        "OFFSET ? ROWS FETCH NEXT ? ROWS ONLY",
        (offset, page_size),
    )
    rows = cursor.fetchall()
    if not rows:
        break
    process_rows(rows)
    offset += page_size

Problemas de transacciones

Alcance de tablas temporales con autocommit

Las tablas temporales de sesión (#tablename) que creas dentro de una transacción desaparecen cuando la transacción se revierte. Este comportamiento suele causar confusión cuando el autocommit está apagado, que es el valor predeterminado:

conn = mssql_python.connect(connection_string)
cursor = conn.cursor()

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")

# An explicit rollback or an error removes #TempData.
conn.rollback()

# This statement fails with "Invalid object name '#TempData'".
cursor.execute("SELECT * FROM #TempData")

Confirma inmediatamente después de crear una tabla temporal, o usa el modo autocommit:

cursor.execute("CREATE TABLE #TempData (ID INT, Name NVARCHAR(50))")
conn.commit()

cursor.execute("INSERT INTO #TempData VALUES (1, 'test')")
conn.commit()

Las sentencias DDL que requieren modo de compromiso automático, como CREATE DATABASE, fallan dentro de una transacción abierta. Establece el autocommit antes de ejecutarlos:

conn.autocommit = True
cursor.execute("CREATE DATABASE TestDB")
conn.autocommit = False

Transacción no comprometida

Síntomas:

Los cambios en los datos no persisten después de cerrar la conexión.

Solution:

Con autocommit=False, que es el valor predeterminado, llama a commit():

cursor.execute("CREATE TABLE #Products (Name NVARCHAR(100))")
cursor.execute(
    "INSERT INTO #Products (Name) VALUES (%(name)s)",
    {"name": "Widget"},
)
conn.commit()

Alternativamente, utiliza el modo de auto-commit:

conn = mssql_python.connect(connection_string, autocommit=True)

Errores de interbloqueo

Síntomas:

OperationalError: [40001] (1205) Transaction ... was deadlocked on lock resources with another process

Solution:

La lógica de reintentos gestiona el fallo inmediato, pero los bloqueos recurrentes indican un problema de diseño. Captura el gráfico de bloqueos y analiza las sentencias y tipos de bloqueo. Las soluciones más comunes incluyen estos cambios:

  • Reordene las operaciones para que las transacciones en conflicto adquieran los bloqueos en la misma secuencia.
  • Reducir el alcance de la transacción.
  • Añadir índices apropiados para reducir la duración de los bloqueos.

Para una guía completa del análisis de bloqueos, consulta la guía de bloqueos. Si usas Azure SQL Database, consulta Analizar y prevenir bloqueos.

Problemas de carga masiva

Infracciones de restricciones durante la copia masiva

Síntomas:

RuntimeError: CHECK constraint ... Conflict occurred in database ...
RuntimeError: Cannot insert duplicate key ... violation of PRIMARY KEY constraint

Causa:

Los datos de tu lote incumplen las restricciones de la tabla, como las restricciones de clave primaria, de unicidad, de comprobación o de clave externa.

Solution:

Valida los datos antes de cargarlos. Para conjuntos de datos grandes, carga los datos en una tabla de staging y luego combínalos en el destino:

cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(100))")
cursor.bulkcopy("##Staging", rows)

# Check for duplicate rows before the merge.
cursor.execute("""
    SELECT s.ID
    FROM ##Staging AS s
    INNER JOIN dbo.Target AS t
        ON s.ID = t.ID
""")
dupes = cursor.fetchall()
if dupes:
    print(f"Skipping {len(dupes)} duplicate rows")

# Insert rows that don't exist in the target.
cursor.execute("""
    INSERT INTO dbo.Target (ID, Name)
    SELECT s.ID, s.Name
    FROM ##Staging AS s
    WHERE NOT EXISTS (
        SELECT 1
        FROM dbo.Target AS t
        WHERE t.ID = s.ID
    )
""")
conn.commit()

Para patrones actualizar/insertar (upsert) con tablas de etapas, consulte Patrones de carga y movimiento de datos.

Errores de mapeo de columnas

Síntomas:

RuntimeError: Bulk copy failure - column count mismatch

Causa:

El número de columnas en tus datos no coincide con el número de columnas de la tabla objetivo, o las columnas están en el orden incorrecto.

Solution:

Asegúrate de que tus datos coincidan con el esquema de la tabla en el orden y la cantidad:

from decimal import Decimal

cursor.execute("""
    SELECT COLUMN_NAME, DATA_TYPE
    FROM INFORMATION_SCHEMA.COLUMNS
    WHERE TABLE_NAME = 'MyTable'
    ORDER BY ORDINAL_POSITION
""")
for col in cursor.fetchall():
    print(col)

rows = [
    (1, "Widget", Decimal("19.99")),
    (2, "Gadget", Decimal("29.99")),
]
cursor.bulkcopy("dbo.MyTable", rows)

Desajustes de tipo durante la copia en bloque

Síntomas:

Los datos se cargan, pero los valores están truncados, redondeados o son incorrectos.

Causa:

Los valores de Python no se corresponden claramente con los tipos de columna destino. Ejemplos comunes incluyen float valores cargados en columnas decimales , que pueden perder precisión, y cadenas sobredimensionadas cargadas en columnas de longitud fija.

Solution:

Usa tipos de Python que coincidan con tu esquema:

from decimal import Decimal

rows = [
    # Use Decimal for decimal and numeric columns.
    (1, "Widget", Decimal("19.99")),
    # Avoid float values because they can lose precision.
    # (1, "Widget", 19.99),
]
cursor.bulkcopy("dbo.Products", rows)

Errores de vinculación de tipos de NumPy

Síntomas:

Los parámetros fallan silenciosamente o generan errores de tipo de datos cuando usas enteros o tipos flotantes NumPy.

Causa:

Tipos de NumPy como numpy.int64 y numpy.int32 no pasan isinstance(x, int) en NumPy 2.x. La inferencia de tipo del conductor no los reconoce, lo que provoca un comportamiento inesperado.

Solution:

Convierte los valores de NumPy a tipos nativos de Python antes de asignarlos:

import numpy as np

cursor.execute(
    "SELECT * FROM Production.Product WHERE ProductID = %(product_id)s",
    {"product_id": int(np.int64(42))},
)

for _, row in df.iterrows():
    cursor.execute(
        "INSERT INTO #Orders (ProductID, Qty) "
        "VALUES (%(product_id)s, %(qty)s)",
        {
            "product_id": int(row["ProductID"]),
            "qty": int(row["Qty"]),
        },
    )

Para conjuntos de datos más grandes, utiliza las rutas de integración de Arrow o pandas . Estas rutas gestionan la conversión de tipos internamente.

Copia masiva con tablas temporales

Síntomas:

cursor.bulkcopy("#TempTable", data) genera RuntimeError: Invalid object name '#TempTable'.

Causa:

bulkcopy() No se pueden resolver las tablas temporales de sesión (#tablename) debido a las limitaciones de búsqueda de metadatos. Las tablas temporales globales (##tablename) y las tablas permanentes funcionan.

Solution:

Utiliza una tabla temporal global o una tabla de ensayo normal:

# A global temp table is visible to all sessions and is dropped
# when the last session disconnects.
cursor.execute("CREATE TABLE ##Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("##Staging", rows)

# Alternatively, use a permanent staging table.
cursor.execute("CREATE TABLE dbo.Staging (ID INT, Name NVARCHAR(50))")
cursor.bulkcopy("dbo.Staging", rows)

Para conjuntos de datos pequeños donde prefieras una tabla temporal de sesión, usa executemany():

cursor.execute("CREATE TABLE #Staging (ID INT, Name NVARCHAR(50))")
cursor.executemany(
    "INSERT INTO #Staging (ID, Name) VALUES (?, ?)",
    rows,
)

Contenido relacionado

  • Last updated on