¿Cómo puedo probar si una consulta de Python SQL funcionó si no se supone que devuelva nada?

He leído un excelente artículo aquí sobre el manejo de errores. https://stackoverflow.com/questions/30996401/error-handling-in-python-mysql

Y hay mil preguntas similares de personas con problemas donde su consulta no devuelve nada, pero eso es porque se supone que no debe hacerlo.

Tengo un problema ligeramente diferente.

Tengo una consulta SQL que está diseñada para transformar la creación de una nueva tabla.

SELECT *
INTO MyDataBase.MySchema.NewTable
FROM

(SELECT ID, Count(*) AS ProductCount FROM MyDataBase.MySchema.MyTable GROUP BY ID ) ProductCount

Si ejecuto la consulta en SSMS, funciona perfectamente y SSMS simplemente me informa:

(1901 filas afectadas)
Tiempo de finalización: 2021-12-07T09:28:32.2541722+13:00

Pero después de deliberadamente mal editar el código y tratar de ejecutar la consulta desde un script de Python, solo me da un resultado vago sin indicar que el script no funcionó en realidad.

fila = curs.fetchone()
pyodbc.ProgrammingError: No hay resultados. La SQL anterior no era una consulta.

Entonces… ¿Cómo puedo verificar que la transformación, la consulta SQL, se haya ejecutado correctamente, incluso si no se supone que debe devolver ningún resultado… desde Python?

Sé que no lo hizo.

davy.ai

July 13, 2023 at 3:53 pm

Para verificar que la consulta SQL se ejecutó correctamente, incluso si no se supone que devuelva resultados, puedes usar el atributo rowcount del objeto cursor en PyODBC. Este atributo devuelve el número de filas afectadas por la última declaración SQL ejecutada utilizando el cursor.

Aquí tienes un ejemplo de fragmento de código que muestra cómo verificar la ejecución exitosa de la consulta SQL:

import pyodbc

cadena_conexion = 'DRIVER={SQL Server Native Client 11.0};SERVER=NombreServidor;DATABASE=MiBaseDeDatos;Trusted_Connection=yes;'
cnxn = pyodbc.connect(cadena_conexion)

cursor = cnxn.cursor()
consulta_sql = """
    SELECT *
    INTO MiBaseDeDatos.MiEsquema.NuevaTabla
    FROM
        (SELECT
            ID,
            Count(*) AS CantidadProductos
        FROM
            MiBaseDeDatos.MiEsquema.MiTabla
        GROUP BY 
            ID
        ) CantidadProductos
"""
try:
    cursor.execute(consulta_sql)
    cantidad_filas_afectadas = cursor.rowcount
    print(f"{cantidad_filas_afectadas} filas afectadas")
except pyodbc.Error as e:
    print(f"Error: {e}")
    # manejar el error aquí

En este fragmento de código, primero creamos una conexión al servidor SQL utilizando PyODBC. Luego creamos un objeto cursor y ejecutamos la consulta SQL utilizando el método execute(). Si la ejecución es exitosa, obtenemos el número de filas afectadas utilizando el atributo rowcount del objeto cursor y lo imprimimos. Si hay un error durante la ejecución, capturamos el error utilizando un bloque try-except y lo manejamos de manera apropiada.

Con este enfoque, puedes verificar la ejecución exitosa de la consulta SQL en Python y manejar cualquier error que pueda ocurrir.

Answer

davy.ai

July 13, 2023 at 3:53 pm
Para verificar que la consulta SQL se ejecutó correctamente, incluso si no se supone que devuelva resultados, puedes usar el atributo rowcount del objeto cursor en PyODBC. Este atributo devuelve el número de filas afectadas por la última declaración SQL ejecutada utilizando el cursor.

Aquí tienes un ejemplo de fragmento de código que muestra cómo verificar la ejecución exitosa de la consulta SQL:
```
import pyodbc

cadena_conexion = 'DRIVER={SQL Server Native Client 11.0};SERVER=NombreServidor;DATABASE=MiBaseDeDatos;Trusted_Connection=yes;'
cnxn = pyodbc.connect(cadena_conexion)

cursor = cnxn.cursor()
consulta_sql = """
    SELECT *
    INTO MiBaseDeDatos.MiEsquema.NuevaTabla
    FROM
        (SELECT
            ID,
            Count(*) AS CantidadProductos
        FROM
            MiBaseDeDatos.MiEsquema.MiTabla
        GROUP BY 
            ID
        ) CantidadProductos
"""
try:
    cursor.execute(consulta_sql)
    cantidad_filas_afectadas = cursor.rowcount
    print(f"{cantidad_filas_afectadas} filas afectadas")
except pyodbc.Error as e:
    print(f"Error: {e}")
    # manejar el error aquí
```
En este fragmento de código, primero creamos una conexión al servidor SQL utilizando PyODBC. Luego creamos un objeto cursor y ejecutamos la consulta SQL utilizando el método execute(). Si la ejecución es exitosa, obtenemos el número de filas afectadas utilizando el atributo rowcount del objeto cursor y lo imprimimos. Si hay un error durante la ejecución, capturamos el error utilizando un bloque try-except y lo manejamos de manera apropiada.

Con este enfoque, puedes verificar la ejecución exitosa de la consulta SQL en Python y manejar cualquier error que pueda ocurrir.

Comments are closed.

es.davy.ai

¿Tienes una pregunta?

¿Cómo puedo probar si una consulta de Python SQL funcionó si no se supone que devuelva nada?

Answer

davy.ai