En esta ocasión, te voy a mostrar como crear un método decorador para la clase BaseDatos, que sea capaz de abrir y cerrar la conexión MySQL desde Python. Así, evitaremos tener que añadirlo a cada método MySQL, los cuales, todos requieren de esto mismo.
Además, te mostraré como aplicar correctamente más de una decoración a un mismo método Python.
El método para eliminar bases de datos
Este método, tiene una complicación extra, está decorado con el método reporte_bd():
#Eliminar bases de datos
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
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.")El método decorador, es este:
#Decorador para el reporte de bases de datos en el servidor
def reporte_bd(funcion_parametro):
def interno(self, *args):
funcion_parametro(self, *args)
print("Estas son las bases de datos que tiene el servidor:")
BaseDatos.mostrar_bd(self)
return internoPor si no recuerdas lo que hacía, era eliminar una base de datos y después, el decorador hacía una llamada al método mostrar_bd, con el que mostraba las bases de datos que quedaban en el servidor.
Esta podría ser una posible solución al cierre de la conexión y el cursor con este método. Sin embargo, presenta problemas debido al decorador.
#Eliminar bases de datos
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
self.cursor.execute(f"DROP DATABASE {nombre_bd}")
print(f"Se eliminó la base de datos {nombre_bd} correctamente.")
except:
# Si ocurre una excepción, se avisa en la consola
print(f"Base de datos '{nombre_bd}' no encontrada.")
finally:
# Cierra el cursor y la conexión
self.cursor.close()
self.conector.close()El problema aquí, es que si utilizamos un bloque finally para cerrar la conexión, como lo hemos hecho con los otros dos métodos del capítulo anterior, tendremos el problema de que el flujo de ejecución, nos jugará malas pasadas. Se cerrará la conexión con el servidor antes de la llamada a las bases de datos y ocurrirá algo como esto (si la base de datos existe):
Resultado en la consola
Se eliminó la base de datos pruebas correctamente.
Estas son las bases de datos que tiene el servidor:
Aquí tienes el listado de las bases de datos del servidor:
No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.
Se cerró la conexión con la base de datos.
O algo como esto (si no existe):
Resultado en la consola
Base de datos 'pruebas' no encontrada.
Estas son las bases de datos que tiene el servidor:
Aquí tienes el listado de las bases de datos del servidor:
No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.
Se cerró la conexión con la base de datos.
Lo que hace aquí, es primero lo del bloque try o except. Después, cierra con el finally la conexión y el cursor y el código que viene posteriormente con el decorador, no funciona correctamente, porque la conexión se ha cerrado y la necesita para mostrar las bases de datos.
Para solucionar este fallo, vamos a crear una decoradora más, que sirva para cerrar la conexión con el servidor y el cursor, así, en cada método que tengamos, no tendremos que estar repitiendo el cierre.
Método decorador para cerrar la conexión MySQL desde Python
Empecemos a crear este método decorador para cerrar la conexión MySQL desde Python.
Este método, primero, probará de hacer la llamada al método externo (el que decoremos). Después de hacerlo, tiene el bloque finally que cierra
# Decorador para el cierre del cursor y la base de datos
def cierre(funcion_parametro):
def interno(self, *args, **kwargs):
try:
# Se llama a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Se informa de un error en la llamada
print("Ocurrió un error.")
finally:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con el servidor.")
return internoProbemos de decorar mostrar_bd(). Es importante, que retires todos los cierres de los métodos que vayas a decorar con este nuevo, para evitar duplicidad en el cierre.
@cierre
def mostrar_bd(self):
try:
# Se informa de que se están obteniendo las bases de datos
print("Aquí tienes el listado de las bases de datos del servidor:")
# Realiza la consulta para mostrar las bases de datos
self.cursor.execute("SHOW DATABASES")
resultado = self.cursor.fetchall()
# Recorre los resultados y los muestra por pantalla
for bd in resultado:
print(f"-{bd[0]}.")
except:
# Si ocurre una excepción, se avisa en la consola
print("No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.")Llamémoslo:
base_datos.mostrar_bd()
Resultado en la consola
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
¡Bien! Está funcionando.
Volvamos al método eliminar_bd(), el cual, está ya decorado por el otro método decorador que tenemos. No hay problema, podemos utilizar decoradores en serie.
¿En qué orden se ejecutan los decoradores en Python?
El orden de ejecución de los decoradores en Python, es a la inversa del flujo de ejecución normal. El primero que se encuentra, es el último en ejecutarse.
Pues bien, vamos a decorar el método eliminar_bd(). Los decoradores son estos dos:
# Decorador para el reporte de bases de datos en el servidor
def reporte_bd(funcion_parametro):
def interno(self, *args):
funcion_parametro(self, *args)
BaseDatos.mostrar_bd(self)
return interno
# Decorador para el cierre del cursor y la base de datos
def cierre(funcion_parametro):
def interno(self, *args, **kwargs):
try:
# Se llama a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Se informa de un error en la llamada
print("Ocurrió un error.")
finally:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con el servidor.")
return interno#Eliminar bases de datos
@cierre
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
self.cursor.execute(f"DROP DATABASE {nombre_bd}")
print(f"Se eliminó la base de datos {nombre_bd} correctamente.")
except:
# Si ocurre una excepción, se avisa en la consola
print(f"Base de datos '{nombre_bd}' no encontrada.")En este caso, primero ejecutará reporte_bd. Esto tiene una llamada a la función externa. Por lo tanto, lo primero que hace, es eliminar o intentar de eliminar la base de datos pasada como argumento. Después, el decorador @reporte_bd, lo que hace es llamar al método mostrar_bd. El método mostrar_bd, está a la vez decorado con @cierre, lo que hace que se cierre la conexión y el cursor, mostrando el mensaje de "Se cerró la conexión con el servidor.".
El problema es que en el método eliminar_bd, necesitamos también un @cierre, entonces, por eso, se muestra duplicado el mensaje.
Resultado en la consola
Se eliminó la base de datos pruebas correctamente.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
Se cerró la conexión con el servidor.
Control del estado de la conexión con el servidor
Para evitar que se intenten cerrar conexiones de manera repetida allá donde decoremos con @cierre, podemos implementar un sistema de control del estado de la conexión con el servidor.
Vamos al decorador @cierre.
# Decorador para el cierre del cursor y la base de datos
def cierre(funcion_parametro):
def interno(self, *args, **kwargs):
try:
# Se llama a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Se informa de un error en la llamada
print("Ocurrió un error.")
finally:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con el servidor.")
return internoVamos a añadir en el método __init__, una variable de control booleana, que establezca un valor para determinar que la conexión con el servidor está activa o cerrada.
En un principio, el valor es False, que indica que la conexión no está cerrada, ya que la iniciamos en el __init__.
self.conexion_cerrada = False # Variable de control para el decorador cierre
Después, en el decorador, hay que añadir en el bloque finally, un condicional if else. Si la conexión está cerrada (conexion_cerrada == True), entonces, no hagas nada en este finally. En cambio, si aún no se ha cerrado, ciérrala.
# Decorador para el cierre del cursor y la base de datos
def cierre(funcion_parametro):
def interno(self, *args, **kwargs):
try:
# Se llama a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Se informa de un error en la llamada
print("Ocurrió un error.")
finally:
if self.conexion_cerrada:
pass
else:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
self.conexion_cerrada = True
print("Se cerró la conexión con el servidor.")
return internoAhora si, si utilizamos decoradores que interfieran con el tema de la conexión, no tendremos más problemas. Eso es lo que podríamos pensar, pero sí. Los hay.
De momento, probemos de llamar al método eliminar_bd().
base_datos.eliminar_bd("pruebas")En este caso, se ha conseguido eliminar la duplicidad del cierre.
Resultado en la consola
Se eliminó la base de datos pruebas correctamente.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
El problema es que como se está realizando la clase, se abre la conexión con el método __init__, pero en el momento en el que cerramos con algún método como eliminar_bd(), ya no se vuelve a abrir. Entonces, solo se puede usar un método correctamente en tiempo de ejecución. Un verdadero problema.
Afortunadamente, hay solución. Voy a modificar el método de cierre, lo llamaré a partir de ahora "conexion" y lo que voy a hacer, es que antes de llamar a un método externo, abra la conexión y que luego la cierre, después de llamar al método externo. De esta forma, de una tenemos todo lo que necesitamos.
En el método __init__, vamos a añadir dos atributos más, para poder usar por toda la clase, el valor del host del servidor y el nombre de usuario. El de la contraseña, ya lo teníamos.
class BaseDatos:
#Conexión y cursor
def __init__(self, **kwargs):
self.conector = mysql.connector.connect(**kwargs)
self.cursor = self.conector.cursor()
self.host = kwargs["host"]
self.usuario = kwargs["user"]
self.contrasena = kwargs["password"]
self.conexion_cerrada = False # Variable de control para el decorador cierre y aperturaAhora que tenemos los nuevos atributos, vamos a modificar el método decorador que antes se llamaba "cierre".
Nos queda un decorador enorme, pero muy útil para no repetir apertura y cierre de la conexión en cada método de la clase.
En el bloque try, si la conexión está cerrada (conexion_cerrada == True), realiza la conexión con los atributos del __init__.
El controlador de la conexión se pone en False, que indica que la conexión no está cerrada.
Después de esto, sale del if y llama a la función externa.
Si lo anterior no ha sido posible porque por ejemplo, hacemos una llamada incorrecta a algún método decorado, avisará de que ocurrió un error en el except.
Sea como sea, nos interesa que se compruebe si la conexión está abierta o cerrada y que la cierre en caso de estar abierta, independientemente de si hay un error en la llamada o si no.
Si la conexión está cerrada (conexion == True) tiene un pass, no hace nada. En cambio, si no es así, cierra la conexión.
# Decorador para la apertura y cierre de la base de datos
def conexion(funcion_parametro):
def interno(self, *args, **kwargs):
try:
if self.conexion_cerrada:
self.conector = mysql.connector.connect(
host = self.host,
user = self.usuario,
password = self.contrasena
)
self.cursor = self.conector.cursor()
self.conexion_cerrada = False
print("Se abrió la conexión con el servidor.")
# Llamamos a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Si ocurre una excepción, se avisa en la consola
print("Ocurrió un error.")
finally:
if self.conexion_cerrada:
pass
else:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
self.conexion_cerrada = True
print("Se cerró la conexión con el servidor.")
return internoVamos a decorar todos los métodos. Puesto que en el punto que estamos ya hay mucha complejidad, te dejo todo el código hasta el momento, por si tienes errores y necesitas verlo completo:
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.host = kwargs["host"]
self.usuario = kwargs["user"]
self.contrasena = kwargs["password"]
self.conexion_cerrada = False # Variable de control para el decorador cierre y apertura
# Decorador para el reporte de bases de datos en el servidor
def reporte_bd(funcion_parametro):
def interno(self, *args):
funcion_parametro(self, *args)
BaseDatos.mostrar_bd(self)
return interno
# Decorador para la apertura y cierre de la base de datos
def conexion(funcion_parametro):
def interno(self, *args, **kwargs):
try:
if self.conexion_cerrada:
self.conector = mysql.connector.connect(
host = self.host,
user = self.usuario,
password = self.contrasena
)
self.cursor = self.conector.cursor()
self.conexion_cerrada = False
print("Se abrió la conexión con el servidor.")
# Llamamos a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Si ocurre una excepción, se avisa en la consola
print("Ocurrió un error.")
finally:
if self.conexion_cerrada:
pass
else:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
self.conexion_cerrada = True
print("Se cerró la conexión con el servidor.")
return interno
# Método para realizar consultas SQL
@conexion
def consulta(self, sql):
try:
# Realiza la consulta o modificación de la base de datos
self.cursor.execute(sql)
print("Esta es la salida de la instrucción que has introducido:")
print(self.cursor.fetchall())
except:
# Si ocurre una excepción, se avisa en la consola
print("Ocurrió un error. Revisa la instrucción SQL.")
finally:
# Cierra el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con la base de datos.")
# Mostrar bases de datos del servidor
@conexion
def mostrar_bd(self):
try:
# Se informa de que se están obteniendo las bases de datos
print("Aquí tienes el listado de las bases de datos del servidor:")
# Realiza la consulta para mostrar las bases de datos
self.cursor.execute("SHOW DATABASES")
resultado = self.cursor.fetchall()
# Recorre los resultados y los muestra por pantalla
for bd in resultado:
print(f"-{bd[0]}.")
except:
# Si ocurre una excepción, se avisa en la consola
print("No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.")
# Eliminar bases de datos
@conexion
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
self.cursor.execute(f"DROP DATABASE {nombre_bd}")
print(f"Se eliminó la base de datos {nombre_bd} correctamente.")
except:
# Si ocurre una excepción, se avisa en la consola
print(f"Base de datos '{nombre_bd}' no encontrada.")
# Crear bases de datos
@conexion
@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
@conexion
def copia_bd(self, nombre_bd):
#Obtiene la fecha y hora actual
fecha_hora = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
with open(f'{carpeta_respaldo}/{nombre_bd}_{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)
# Crear tablas en una base de datos
@conexion
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 adminte 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()
@conexion
def eliminar_tabla(self, nombre_bd, nombre_tabla):
self.cursor.execute(f"USE {nombre_bd}")
self.cursor.execute(f"DROP TABLE {nombre_tabla}")A partir de esto, empecemos a probar los métodos, a ver si funcionan como es de esperar o hay que hacerles alguna mejora.
Método de consultas SQL
Empecemos por el método consulta.
# Método para realizar consultas SQL
@conexion
def consulta(self, sql):
try:
# Realiza la consulta o modificación de la base de datos
self.cursor.execute(sql)
print("Esta es la salida de la instrucción que has introducido:")
print(self.cursor.fetchall())
except:
# Si ocurre una excepción, se avisa en la consola
print("Ocurrió un error. Revisa la instrucción SQL.")
finally:
# Cierra el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con la base de datos.")Le tenemos que retirar el bloque finally, ya no necesitamos el cierre ni que nos lo indique, de eso se encarga el decorador.
# Método para realizar consultas SQL
@conexion
def consulta(self, sql):
try:
# Realiza la consulta o modificación de la base de datos
self.cursor.execute(sql)
print("Esta es la salida de la instrucción que has introducido:")
print(self.cursor.fetchall())
except:
# Si ocurre una excepción, se avisa en la consola
print("Ocurrió un error. Revisa la instrucción SQL.")Llamo al método con una consulta cualquiera:
base_datos.consulta("SELECT * FROM world.city LIMIT 5")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)]
Se cerró la conexión con el servidor.
El decorador me avisa de que se cerró la conexión, pero no de que se abrió. Esto es porque en el __init__ ya se ha abierto.
Si quieres, puedes añadirle al __init__ un mensaje, para ver este mensaje. Piensa que los mensajes de apertura y cierre del decorador, son para la fase de desarrollo, después, con el producto final, al usuario no le interesa saber si se abrió o se cerró.
Lo pongo en la línea 10.
#Conexión y cursor
def __init__(self, **kwargs):
self.conector = mysql.connector.connect(**kwargs)
self.cursor = self.conector.cursor()
self.host = kwargs["host"]
self.usuario = kwargs["user"]
self.contrasena = kwargs["password"]
self.conexion_cerrada = False # Variable de control para el decorador cierre y apertura
# Avisa de que se abrió la conexión con el servidor
print("Se abrió la conexión con el servidor.")Al hacer la llamada, sale el mensaje.
Resultado en la consola
Se abrió la conexión con el servidor.
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)]
Se cerró la conexión con el servidor.
Ahora, probemos de llamar dos veces al método:
base_datos.consulta("SELECT * FROM world.city LIMIT 5")
base_datos.consulta("SELECT * FROM world.city LIMIT 2")Con la primera llamada, se abre la conexión, ejecuta el código del método consulta y se cierra. Para la siguiente llamada, la vuelve a abrir, ejecuta el código del método y la cierra de nuevo. Está funcionando.
Resultado en la consola
Se abrió la conexión con el servidor.
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)]
Se cerró la conexión con el servidor.
Se abrió la conexión con el servidor.
Esta es la salida de la instrucción que has introducido:
[(1, 'Kabul', 'AFG', 'Kabol', 1780000), (2, 'Qandahar', 'AFG', 'Qandahar', 237500)]
Se cerró la conexión con el servidor.
Si se produce un error en la llamada, por ejemplo, dejándola sin argumentos, nos dirá que ocurrió un error. Esto lo saca en la consola el propio método decorador con su except:
base_datos.consulta()
La conexión se abre, se intenta ejecutar la consulta, pero ocurrió un error. Acto seguido, cierra la conexión.
Resultado en la consola
Se abrió la conexión con el servidor.
Ocurrió un error.
Se cerró la conexión con el servidor.
Pasemos al siguiente método, mostrar_bd().
Método para mostrar bases de datos desde Python
El método para mostrar bases de datos, mostrar_bd(), lo tengo así:
# Mostrar bases de datos del servidor
@conexion
def mostrar_bd(self):
try:
# Se informa de que se están obteniendo las bases de datos
print("Aquí tienes el listado de las bases de datos del servidor:")
# Realiza la consulta para mostrar las bases de datos
self.cursor.execute("SHOW DATABASES")
resultado = self.cursor.fetchall()
# Recorre los resultados y los muestra por pantalla
for bd in resultado:
print(f"-{bd[0]}.")
except:
# Si ocurre una excepción, se avisa en la consola
print("No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.")Probemos la salida de una llamada ahora que está decorado.
base_datos.mostrar_bd()
También funciona como era de esperar.
Resultado en la consola
Se abrió la conexión con el servidor.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-pruebas.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
Método para eliminar bases de datos desde Python
El método para eliminar bases de datos, es un poco más complicado, ya que está decorado con dos decoradores.
# Eliminar bases de datos
@conexion
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
self.cursor.execute(f"DROP DATABASE {nombre_bd}")
print(f"Se eliminó la base de datos {nombre_bd} correctamente.")
except:
# Si ocurre una excepción, se avisa en la consola
print(f"Base de datos '{nombre_bd}' no encontrada.")Hacemos la llamada:
base_datos.eliminar_bd("pruebas")Resultado en la consola
Se abrió la conexión con el servidor.
Se eliminó la base de datos pruebas correctamente.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
En este caso, se inicia la conexión desde el __init__ ("Se abrió la conexión con el servidor.").
Se ejecuta el método eliminar_bd(). Este, ejecuta el código del decorador reporte_bd().
El decorador reporte_bd() ejecuta la llamada al método mostrar_bd(), con lo que se muestra el listado de bases de datos. Finalmente, actua el último decorador, el de la conexión, cerrándola.
Si la base de datos no existe, esta es la salida:
Resultado en la consola
Se abrió la conexión con el servidor.
Base de datos 'pruebas' no encontrada.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
Otro método que está listo.
El método para crear bases de datos desde Python
El siguiente método es el de crear bases de datos, crear_bd().
# Crear bases de datos
@conexion
@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}.")Este también está decorado por los dos decoradores.
base_datos.crear_bd("pruebas")Resultado en la consola
Se abrió la conexión con el servidor.
Se creó la base de datos pruebas o ya estaba creada.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-pruebas.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
La salida funciona bien, esté creada o no la base de datos.
En caso de error, por ejemplo, introduciendo un carácter no válido en el nombre del argumento, salta el except del método crear_bd().
base_datos.crear_bd("pruebas.base.datos")Resultado en la consola
Se abrió la conexión con el servidor.
Ocurrió un error al intentar crear la base de datos pruebas.base.datos.
Aquí tienes el listado de las bases de datos del servidor:
-information_schema.
-mysql.
-performance_schema.
-pruebas.
-sakila.
-sys.
-world.
Se cerró la conexión con el servidor.
Método para hacer copias de seguridad MySQL desde Python
Veamos que ocurre al hacer una copia de seguridad MySQL con el método copia_bd() decorado con la conexión.
#Crear backups de bases de datos
@conexion
def copia_bd(self, nombre_bd):
#Obtiene la fecha y hora actual
fecha_hora = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
with open(f'{carpeta_respaldo}/{nombre_bd}_{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) base_datos.copia_bd("world")Resultado en la consola
Se abrió la conexión con el servidor.
Se cerró la conexión con el servidor.
Todo funciona, sin embargo, estaría bien recibir una confirmación de que la copia se pudo realizar o no.
Modificamos el método de copias con un try - except:
# En caso de error al crear la copia, nos avisa en la consola
except:
print("Ocurrió un error al intentar crear la copia de seguridad.") Sin embargo, este except no va a funcionar, por ejemplo, con un error con mysqldump con el nombre de la base de datos:
Error en la consola
Se abrió la conexión con el servidor.
Se cerró la conexión con el servidor.
Got error: 1049: Unknown database 'nombre_bd_que_no_existe' when selecting the database
Esto tiene algo de complicación para manejar dicha excepción, así que para no darte más y más datos en este capítulo, haremos que simplemente, se realice una comprobación de si la base de datos existe antes de intentar crear la copia. Así evitamos el error de base de datos inexistente de mysqldump.
En la variable sql, se almacena una instrucción SQL. Esta instrucción es la de mostrar bases de datos, solo que añadiéndole la cláusula SQL "LIKE" (como) buscamos una cadena de texto que cumpla un patrón, en este caso, sencillamente el nombre de la base de datos.
En resumen, la instrucción SQL, busca una base de datos que sea igual al nombre proporcionado en la llamada del método copia_bd().
El cursor ejecuta esa instrucción. Con fetchone(), como ya expliqué, almacenamos el primer resultado de la consulta SQL. Lo que se guarda en "resultado", es una tupla con una posición en el caso de que la base de datos exista en el servidor MySQL (al igual que el listado de mostrar_bd(), pero solo una base de datos, la que le pasamos como argumento al método copia_bd()), en caso contrario, almacena un valor None.
#Crear backups de bases de datos
@conexion
def copia_bd(self, nombre_bd):
# Verifica si la base de datos existe en el servidor
sql = f"SHOW DATABASES LIKE '{nombre_bd}'"
self.cursor.execute(sql)
resultado = self.cursor.fetchone()
# Si la base de datos no existe, muestra un mensaje de error y termina la función
if not resultado:
print(f'La base de datos {nombre_bd} no existe.')
return
#Obtiene la fecha y hora actual
fecha_hora = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
with open(f'{carpeta_respaldo}/{nombre_bd}_{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) Probemos esto. Vamos a imprimir el valor de resultado, cuando ponemos un nombre incorrecto o inexistente en el servidor:
base_datos.copia_bd("nombre_bd.que_no_existe")En este caso, nos imprime un None, ya que no existe (es incorrecta en este caso).
Resultado en la consola
Se abrió la conexión con el servidor.
None
La base de datos nombre_bd.que_no_existe no existe.
Se cerró la conexión con el servidor.
En el caso de que si exista, esto es lo que se almacena en la variable resultado:
base_datos.copia_bd("world")Resultado en la consola
Se abrió la conexión con el servidor.
('world',)
Se cerró la conexión con el servidor.
Entonces, en el if, se evalua si resultado es false.
En Python, cuando queremos evaluar todo lo contrario a if nombre_variable (abreviación de if nombre variable == True), podemos poner if not nombre_variable (abreviación de if nombre_variable == False).
None, es uno de los valores "falsy" de Python, tema que no he explicado todavía, pero que sepas, que None, evalúa como False.
Entonces, si la base de datos proporcionada, no existe, se imprime un print(). Finalmente, hay que finalizar antes de tiempo la ejecución del método, para que no intente lo del código que viene a continuación.
Entonces, puedes pensar en poner un break, pero esto es solo para los condicionales o bucles.
Para finalizar la ejecución en un punto determinado de una función, hay que usar un return.
Finalmente, voy a añadir abajo del todo, un print() que salga solo si se ha creado la copia de seguridad (línea 18):
#Crear backups de bases de datos
@conexion
def copia_bd(self, nombre_bd):
# Verifica si la base de datos existe en el servidor
sql = f"SHOW DATABASES LIKE '{nombre_bd}'"
self.cursor.execute(sql)
resultado = self.cursor.fetchone()
# Si la base de datos no existe, muestra un mensaje de error y termina la función
if not resultado:
print(f'La base de datos {nombre_bd} no existe.')
return
#Obtiene la fecha y hora actual
fecha_hora = datetime.datetime.now().strftime("%Y-%m-%d %H-%M-%S")
with open(f'{carpeta_respaldo}/{nombre_bd}_{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)
print("Se creó la copia de seguridad correctamente.")Hagamos un par de pruebas. Si la base de datos existe…
base_datos.copia_bd("world")Resultado en la consola
Se abrió la conexión con el servidor.
Se creó la copia de seguridad correctamente.
Se cerró la conexión con el servidor.
En cambio, si no existe…
base_datos.copia_bd("base_datos_inexistente")Resultado en la consola
Se abrió la conexión con el servidor.
La base de datos base_datos_inexistente no existe.
Se cerró la conexión con el servidor.
Método para crear tablas MySQL desde Python
Pasemos a uno de los métodos más complejos que hay en este proyecto. Probemos si funciona solo decorándolo con el decorador de la conexión.
# Crear tablas en una base de datos
@conexion
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 adminte 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 servidorRealicemos la llamada, para crear una tabla que no está en la base de datos.
base_datos.crear_tabla("pruebas","usuarios", tbl.columnas)Resultado en la consola
Se abrió la conexión con el servidor.
Se cerró la conexión con el servidor.
Aparentemente, está todo correcto. Puedes comprobarlo en Workbench.
Hay que poner un mensaje que confirme la creación de la tabla y otro cuando no se pudo crear. Le añado el try y el except:
@conexion
def crear_tabla(self, nombre_bd, nombre_tabla, columnas):
try:
# 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 adminte 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 confirma
self.conector.commit()
# Se informa de que la creación se ha efectuado correctamente.
print("Se creó la tabla correctamente.")
except:
print("Ocurrió un error al intentar crear la tabla.")Ahora, si la tabla no existe y se puede crear…
base_datos.crear_tabla("pruebas","usuarios2", tbl.columnas)Resultado en la consola
Se abrió la conexión con el servidor.
Se creó la tabla correctamente.
Se cerró la conexión con el servidor.
En cambio, si ya está creada…
base_datos.crear_tabla("pruebas","usuarios", tbl.columnas)Resultado en la consola
Se abrió la conexión con el servidor.
Ocurrió un error al intentar crear la tabla.
Se cerró la conexión con el servidor.
Bueno. Aquí podrías mejorar esto añadiendo un control que compruebe si la tabla existe o no antes de intentar crearla. Esto ya te lo dejo a ti si quieres practicar más.
Vayamos a por el último método, eliminar_bd().
Método para eliminar tablas MySQL desde Python
El último método que nos queda por revisar, es el de eliminar tablas MySQL. Este es uno de los métodos más sencillos.
@conexion
def eliminar_tabla(self, nombre_bd, nombre_tabla):
self.cursor.execute(f"USE {nombre_bd}")
self.cursor.execute(f"DROP TABLE {nombre_tabla}")Antes de probar nada, voy a poner ya el try except para que se avise en la consola de lo que ocurre.
Lo que me gustaría ahora, es que en el except, se mostrara un reporte de las tablas que hay en la base de datos proporcionada como argumento.
@conexion
def eliminar_tabla(self, nombre_bd, nombre_tabla):
try:
self.cursor.execute(f"USE {nombre_bd}")
self.cursor.execute(f"DROP TABLE {nombre_tabla}")
print(f"Tabla '{nombre_tabla}' eliminada correctamente de la base de datos {nombre_bd}.")
except:
print(f"No se pudo eliminar la tabla '{nombre_tabla}' de la base de datos '{nombre_bd}'.")Probamos de eliminar una tabla que existe en la base de datos pruebas:
base_datos.eliminar_tabla("pruebas","usuarios")Resultado en la consola
Se abrió la conexión con el servidor.
Tabla 'usuarios' eliminada correctamente de la base de datos 'pruebas'.
Se cerró la conexión con el servidor.
Si la tabla no existe, entonces, salta el bloque except:
Resultado en la consola
Se abrió la conexión con el servidor.
No se pudo eliminar la tabla 'usuarios' de la base de datos 'pruebas'.
Se cerró la conexión con el servidor.
Aquí tienes todo el temario del curso Máster en Python.
Te dejo por aquí, todo el código del capítulo por si algo no te funciona y lo necesitas para continuar con el siguiente capítulo.
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.host = kwargs["host"]
self.usuario = kwargs["user"]
self.contrasena = kwargs["password"]
self.conexion_cerrada = False
# Avisa de que se abrió la conexión con el servidor
print("Se abrió la conexión con el servidor.")
#Decoradora para el reporte de bases de datos en el servidor
def reporte_bd(funcion_parametro):
def interno(self, nombre_bd):
funcion_parametro(self, nombre_bd)
BaseDatos.mostrar_bd(self)
return interno
# Decorador para el cierre del cursor y la base de datos
def conexion(funcion_parametro):
def interno(self, *args, **kwargs):
try:
if self.conexion_cerrada:
self.conector = mysql.connector.connect(
host = self.host,
user = self.usuario,
password = self.contrasena
)
self.cursor = self.conector.cursor()
self.conexion_cerrada = False
print("Se abrió la conexión con el servidor.")
# Se llama a la función externa
funcion_parametro(self, *args, **kwargs)
except:
# Se informa de un error en la llamada
print("Ocurrió un error con la llamada.")
finally:
if self.conexion_cerrada:
pass
else:
# Cerramos el cursor y la conexión
self.cursor.close()
self.conector.close()
print("Se cerró la conexión con el servidor.")
self.conexion_cerrada = True
return interno
#Consultas SQL
@conexion
def consulta(self, sql):
try:
self.cursor.execute(sql)
print("Esta es la salida de la instrucción que has introducido:")
print(self.cursor.fetchall())
except:
print("Ocurrió un error. Revisa la instrucción SQL.")
@conexion
def mostrar_bd(self):
try:
# Se informa de que se están obteniendo las bases de datos
print("Aquí tienes el listado de las bases de datos del servidor:")
# Realiza la consulta para mostrar las bases de datos
self.cursor.execute("SHOW DATABASES")
resultado = self.cursor.fetchall()
# Recorre los resultados y los muestra por pantalla
for bd in resultado:
print(f"-{bd[0]}.")
except:
# Si ocurre una excepción, se avisa en la consola
print("No se pudieron obtener las bases de datos. Comprueba la conexión con el servidor.")
#Eliminar bases de datos
@conexion
@reporte_bd
def eliminar_bd(self, nombre_bd):
try:
# Realiza la consulta para eliminar la base de datos
self.cursor.execute(f"DROP DATABASE {nombre_bd}")
print(f"Se eliminó la base de datos {nombre_bd} correctamente.")
except:
# Si ocurre una excepción, se avisa en la consola
print(f"Base de datos '{nombre_bd}' no encontrada.")
#Crear bases de datos
@conexion
@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
@conexion
def copia_bd(self, nombre_bd):
# Verifica si la base de datos existe en el servidor
sql = f"SHOW DATABASES LIKE '{nombre_bd}'"
self.cursor.execute(sql)
resultado = self.cursor.fetchone()
print(resultado)
# Si la base de datos no existe, muestra un mensaje de error y termina el método
if not resultado:
print(f'La base de datos {nombre_bd} no existe.')
return
#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)
print("Se creó la copia correctamente.")
@conexion
def crear_tabla(self, nombre_bd, nombre_tabla, columnas):
try:
#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 informa de que la creación se ha efectuado correctamente.
print("Se creó la tabla correctamente.")
except:
print("Ocurrió un error al intentar crear la tabla.")
@conexion
def eliminar_tabla(self, nombre_bd, nombre_tabla):
try:
self.cursor.execute(f"USE {nombre_bd}")
self.cursor.execute(f"DROP TABLE {nombre_tabla}")
print(f"Tabla '{nombre_tabla}' eliminada correctamente de la base de datos {nombre_bd}.")
except:
print(f"No se pudo eliminar la tabla '{nombre_tabla}' de la base de datos '{nombre_bd}'.")
Muy buen tutorial, muy bien explicado. Hay que tener mucho cuidado y paciencia, para uno que es novato como yo; se me complico un poco; pero insistí en escribir el código y no copiar el tuyo hasta que encontraba los pequeños errores de escritura. Muchas gracias por tu enseñanza.
Hola,
En mis pruebas encontré que el método «base_datos.eliminar_bd(«pruebas) no funciona cuando el decorador está ANIDADO con «@conexion» y «@reporte_bd» «. En el video aparentemente si funciona, pero es porque si se observa el minuto 17:55 la base de datos «pruebas» no existe.
NOTA: al quitar el anidamiento y dejar solo el decorador «@conexion» si funciona.
Disculpe, pero no entiendo porque desde que agregue los decoradores según lo indicado en el material de día #38, siempre me envía el siguiente mensaje cuando trato de ejecutar cualquiera de los métodos… el siguiente es el ejemplo de error que me envía cuando trato de ejecutar mostrar_bd()…
c:\Users\clori\Desktop\APP_CONSOLA_CRUD\PROYECTO-BD\bd\respaldo
….init….Se abrio la conexion con el servidor????….
Traceback (most recent call last):
File «c:\Users\clori\Desktop\APP_CONSOLA_CRUD\PROYECTO-BD\xxapp.py», line 14, in
xbdatos.mostrar_bd()
TypeError: ‘NoneType’ object is not callable
Como puedo corregir este error?