El cursor de MySQL Connector, tiene tres métodos muy útiles para trabajar con consultas desde Python. Así pues, en este capítulo veremos su uso aplicado al proyecto de Python que estamos realizando en el curso.
Los métodos son fetchall, fetchone y fetchmany.
Seguimos añadiendo cosas al proyecto de bases de datos. Para asegurarme de que todo el mundo puede seguir el hilo, dejo aquí todo el código que llevamos por el momento.
Estado del proyecto
El estado del proyecto en este capítulo es el siguiente:
app.py
#Importaciones import bd.base_datos as sqlbd import bd.tablas as tbl base_datos = sqlbd.BaseDatos(**sqlbd.acceso_bd)
base_datos.py
import mysql.connector import os import subprocess import datetime #conexion con la base de datos acceso_bd = {"host" : "localhost", "user" : "root", "password" : "programacionfacil", } # --> Rutas #Obtenemos la raíz de la carpeta del proyecto carpeta_principal = os.path.dirname(__file__) carpeta_respaldo = os.path.join(carpeta_principal, "respaldo") class BaseDatos: #Conexión y cursor def __init__(self, **kwargs): self.conector = mysql.connector.connect(**kwargs) self.cursor = self.conector.cursor() self.contrasena = kwargs["password"] #Decoradora para el reporte de bases de datos en el servidor def reporte_bd(funcion_parametro): def interno(self, nombre_db): funcion_parametro(self, nombre_db) print("Estas son las bases de datos que tiene el servidor:") BaseDatos.mostrar_bd(self) return interno #Consultas SQL def consulta(self, sql): self.cursor.execute(sql) return self.cursor #Mostrar bases de datos def mostrar_bd(self): self.cursor.execute("SHOW DATABASES") for bd in self.cursor: print(bd) #Eliminar bases de datos @reporte_bd def eliminar_bd(self, nombre_bd): try: self.cursor.execute(f"DROP DATABASE {nombre_bd}") print(f"Se eliminó la base de datos {nombre_bd} correctamente.") except: print(f"Base de datos '{nombre_bd}' no encontrada.") #Crear bases de datos @reporte_bd def crear_bd(self, nombre_bd): try: self.cursor.execute(f"CREATE DATABASE IF NOT EXISTS {nombre_bd}") print(f"Se creó la base de datos {nombre_bd} o ya estaba creada.") except: print(f"Ocurrió un error al intentar crear la base de datos {nombre_bd}.") #Crear backups de bases de datos def copia_bd(self, nombre_bd): #Obtiene la hora y fecha actuales self.fecha_hora = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S") #Se crea la copia de seguridad with open(f'{carpeta_respaldo}/{nombre_bd}_{self.fecha_hora}.sql', 'w') as out: subprocess.Popen(f'"C:/Program Files/MySQL/MySQL Workbench 8.0/"mysqldump --user=root -- password={self.contrasena} --databases {nombre_bd}', shell=True, stdout=out) def crear_tabla(self, nombre_bd, nombre_tabla, columnas): #String para guardar el string con las columnas y tipos de datos columnas_string = "" #Se itera la lista que se le pasa como argumento (cada diccionario) for columna in columnas: #formamos el string con nombre, tipo y longitud columnas_string += f"{columna['name']} {columna['type']}({columna['length']})" #Si es clave primaria, auto_increment o no admite valores nulos, lo añade al string if columna['primary_key']: columnas_string += " PRIMARY KEY" if columna['auto_increment']: columnas_string += " AUTO_INCREMENT" if columna['not_null']: columnas_string += " NOT NULL" #Hace un salto de línea después de cada diccionario columnas_string += ",\n" #Elimina al final del string el salto de línea y la coma columnas_string = columnas_string[:-2] #Le indica que base de datos utilizar self.cursor.execute(f"USE {nombre_bd}") #Se crea la tabla juntando la instrucción SQL con el string generado sql = f"CREATE TABLE {nombre_tabla} ({columnas_string});" #Se ejecuta la instrucción self.cursor.execute(sql) #Se hace efectiva self.conector.commit() #Se cierra la conexión con el servidor self.conector.close() #Eliminar tablas en las bases de datos def eliminar_tabla(self, nombre_bd, nombre_tabla): self.cursor.execute(f"USE {nombre_bd}") self.cursor.execute(f"DROP TABLE {nombre_tabla}")
En este capítulo, volvemos con el proyecto. Esta vez, empezaremos a mejorar los métodos que ya tenemos.
Si lo tienes todo funcionando correcto, empecemos por el más fácil de todos, el método consulta.
Método para consultas SQL
El método consulta que tenemos en el proyecto, tiene una funcionalidad muy limitada. Ya que realmente, no hace prácticamente nada. Lo puse a modo de introducción en la clase.
#Consultas SQL def consulta(self, sql): self.cursor.execute(sql) return self.cursor
Vamos a mejorarlo para que pueda ejecutar cualquier tipo de instrucción SQL y que nos dé un reporte sencillo en la consola.
Normalmente, lo que se necesita cuando trabajamos con el cursor, es que se recorra con él todo el resultado. Algo como lo que hemos hecho anteriormente con el método mostrar_bd().
Pues bien, en el módulo mysql.connector, tenemos el método fetchall() del cursor, que permite recorrer todo con el cursor y devuelve tuplas (una por cada fila del resultado).
Ten en cuenta, que hay otros métodos como fetchone() que solo muestra un resultado y fetchmany() que muestra un número limitado de ellos.
Estos dos métodos son necesarios, si tenemos grandes bases de datos y queremos evitar un consumo desmesurado de recursos al poder devolver absolutamente todos los resultados con fetchall().
Ahora te mostraré unos ejemplos.
El método MySQL Python fetchall()
El método fetchall() del cursor, como he indicado, es para devolver todas las filas en una consulta SQL, pues eso es lo que va a imprimir en la consola a raíz de la instrucción SQL que le pasemos como argumento.
#Ejecuta y muestra resultados de instrucciones SQL def consulta(self, sql): self.cursor.execute(sql) print("Esta es la salida de la instrucción que has introducido:") print(self.cursor.fetchall())
base_datos.consulta("SHOW DATABASES")
Resultado en la consola
Esta es la salida de la instrucción que has introducido:
[('information_schema',), ('mysql',), ('performance_schema',), ('pruebas',), ('sakila',), ('sys',), ('world',)]
Este método, ahora sirve para pasarle la instrucción SQL que quieras, vamos a probar con un SELECT.
base_datos.consulta("SELECT * FROM world.city;")
Resultado en la consola
[(1, 'Kabul', 'AFG', 'Kabol', 1780000), (2, 'Qandahar', 'AFG', 'Qandahar', 237500), (3, 'Herat', 'AFG', 'Herat', 186800), (4, 'Mazar-e-Sharif', 'AFG', 'Balkh', 127800), (5, 'Amsterdam', 'NLD', 'Noord-Holland', 731200),...
Este resultado que te muestro, es de solo 6 filas o registros. Pero la consulta ha devuelto en realidad ¡más de 4000!
Si quieres evitar que tus usuarios puedas hacer uso excesivo de recursos, deberás cambiar el fetchall() por alguno de los otros dos métodos.
El método MySQL Python fetchone()
El método fetchone() sirve para que se devuelva únicamente una sola fila. Independientemente de si la consulta tiene 100 o 1000 registros.
#Ejecuta y muestra resultados de instrucciones SQL def consulta(self, sql): self.cursor.execute(sql) print("Esta es la salida de la instrucción que has introducido:") print(self.cursor.fetchone())
base_datos.consulta("SELECT * FROM world.city;")
Resultado en la consola
[(1, 'Kabul', 'AFG', 'Kabol', 1780000)
Claro, en algunas ocasiones, nos puede servir, pero para este tipo de consultas, no mucho. Es aquí, donde podemos usar un método intermedio que no lo saque todo, pero que tampoco saque un único registro.
El método MySQL Python fetchmany()
El método fetchmany() recibe como argumento, un número cualquiera de filas máximas al recorrer consultas. Por ejemplo, queremos que máximo salgan los 10 primeros registros de una consulta, pues haz esto:
#Ejecuta y muestra resultados de instrucciones SQL def consulta(self, sql): self.cursor.execute(sql) print("Esta es la salida de la instrucción que has introducido:") print(self.cursor.fetchmany(10))
Resultado en la consola
Esta es la salida de la instrucción que has introducido:
[(1, 'Kabul', 'AFG', 'Kabol', 1780000), (2, 'Qandahar', 'AFG', 'Qandahar', 237500), (3, 'Herat', 'AFG', 'Herat', 186800), (4, 'Mazar-e-Sharif', 'AFG', 'Balkh', 127800), (5, 'Amsterdam', 'NLD', 'Noord-Holland', 731200), (6, 'Rotterdam', 'NLD', 'Zuid-Holland', 593321), (7, 'Haag', 'NLD', 'Zuid-Holland', 440900), (8, 'Utrecht', 'NLD', 'Utrecht', 234323), (9, 'Eindhoven', 'NLD', 'Noord-Brabant', 201843), (10, 'Tilburg', 'NLD', 'Noord-Brabant', 193238)]
Se puede también, dejar la opción de pasar el valor del fetchmany como argumento. De esta forma, podremos controlar en cada llamada, el número de registros que queremos en la salida.
Aquí, le doy con un parámetro posicional, un valor por defecto de 10. Ese será el que obtendrá fetchmany() si no le especificamos otra cosa.
#Ejecuta y muestra resultados de instrucciones SQL def consulta(self, sql, registros = 10): self.cursor.execute(sql) print("Esta es la salida de la instrucción que has introducido:") print(self.cursor.fetchmany(registros))
base_datos.consulta("SELECT * FROM world.city;", registros=7)
Resultado en la consola
Esta es la salida de la instrucción que has introducido:
[(1, 'Kabul', 'AFG', 'Kabol', 1780000), (2, 'Qandahar', 'AFG', 'Qandahar', 237500), (3, 'Herat', 'AFG', 'Herat', 186800), (4, 'Mazar-e-Sharif', 'AFG', 'Balkh', 127800), (5, 'Amsterdam', 'NLD', 'Noord-Holland', 731200), (6, 'Rotterdam', 'NLD', 'Zuid-Holland', 593321), (7, 'Haag', 'NLD', 'Zuid-Holland', 440900)]
Por cierto, desde SQL puedes limitar las consultas con la cláusula LIMIT:
base_datos.consulta("SELECT * FROM world.city LIMIT 3")
Resultado en la consola
Esta es la salida de la instrucción que has introducido:
[(1, 'Kabul', 'AFG', 'Kabol', 1780000), (2, 'Qandahar', 'AFG', 'Qandahar', 237500), (3, 'Herat', 'AFG', 'Herat', 186800)]
En este caso, el que manda es SQL en el propio servidor, de modo, que por mucho fetchall o lo que sea que tengamos, si limitamos desde SQL, al iterador de los métodos fetch, les llega ya el límite establecido, ya no reciben todos los registros.
En el siguiente capítulo, comenzaremos con la optimización de la clase BaseDatos, ya que está habiendo muchos métodos que abren conexiones con el servidor y cursores y nunca se cierran.
Aquí tienes todos los enlaces al material completo del curso de Máster en Python.