6.2. MySQL

Nota

Propósito: El motor de base de datos relacionales MySQL, tiene un adaptador Python e implementa la especificación DB API v2.0 (PEP-249), el objeto de esta guía es para explicar y demostrar como usarlo como desarrollador.

../_images/mysql_textlogo.png

Figura 6.6, Logotipo de MySQL

El módulo PyMySQL, es el adaptador de base de datos MySQL para el lenguaje de programación Python. Sus principales características son la implementación de la especificación Python DB-API 2.0 y que esta desarrollado puramente en Python.

A diferencia del módulo SQLite, no hay un módulo Python SQL predeterminado en la librería estándar de Python, que pueda usar para conectarse a una base de datos MySQL. En su lugar, deberá instalar un módulo externor que funja como controlador Python SQL para el servidor MySQL para poder interactuar con base de datos desde aplicaciones de Python.

Truco

Es un adaptador de base de datos MySQL para el lenguaje de programación Python.

6.2.1. Instalación

Para conectarte al servidor MySQL necesita el módulo PyMySQL. Esto significa que debe instalar PyMySQL ejecutando los siguientes comandos correspondiente a cada sistema operativo, los cuales se presentan a continuación:

Para trabajar una aplicación con bases de datos relacionales MySQL requiere instalar las siguientes librerías/módulos:

  1. Entorno de desarrollo.

  2. Python package installer - pip.

  3. Entorno virtual Python.

  4. Instalar el módulo PyMySQL, ejecutando el siguiente comando:

    pip3 install PyMySQL

  5. Motor de base de datos MySQL.

Puede probar si la instalación se realizo correctamente, ejecutando el siguiente comando correspondiente a tu sistema operativo:

python3 -c "import pymysql ; print(pymysql.__version__)"

Si muestra el número de la versión instalada de PyMySQL, tiene correctamente instalado el módulo. Con esto, ya tiene todo listo para continuar.

6.2.1.1. Servidor MySQL

Para instalar el servidor MySQL existen varias formas de realizarlo, para en este caso se realizara con la tecnología Docker. Esto significa que debe instalar en tu sistema operativo:

Luego de instalar las herramientas necesarias, debe ejecutar el siguiente comando correspondiente:

docker run -d --name mysql -e MYSQL_ROOT_PASSWORD=root -e MYSQL_DATABASE=sistema -p 3306:3306 -v my_data:/var/lib/mysql --restart always mysql:latest

El comando anterior crea un contenedor Docker llamado mysql con la version latest, ejecutándose en el puerto 3306 con la base de datos llamada sistema e incluye un punto de montaje my_data.

De esta forma ha instalado y ejecutado el servidor MySQL necesario para las próximas script Python a ejecutar. Con esto, ya tiene todo listo para continuar.

6.2.1.1.1. Estructura de archivos

Para crear la estructura de archivos del proyecto MySQL debe ejecutar los siguientes comandos:

Crear el directorio crud con el siguiente comando:

Crear y acceder al directorio en un solo comando, ejecutando el siguiente comando:

mkdir -p ~/proyectos/mysql/crud && cd $_

El comando anterior crea la siguiente estructura de directorios:

proyectos/
└── mysql/
    └── crud/

Si tiene la estructura de archivo previa, entonces puede continuar con la siguiente sección.

6.2.2. Cadenas de conexión

Para definir el método connect debe definir las cadenas de conexión con MySQL como se describe a continuación:

USER

Usuario de conexión a la base de datos.

PASSW

Contraseña del usuario de conexión a la base de datos.

HOST

IP o dirección DNS de conexión al servidor de la base de datos.

PORT

Puerto de conexión al servidor de la base de datos, por defecto es 3306.

DB

Nombre de la base de datos a cual conectar.

A continuación presento un ejemplo en Python implementando una cadena de conexión para una base de datos MySQL:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
import pymysql

USER = "root"
PASSW = "root"
HOST = "localhost"
PORT = 3306
DB = "sistema"

conexion_bd = pymysql.connect(
    user=USER, password=PASSW, host=HOST, port=PORT, database=DB
)

El ejemplo anterior se describe a continuación:

  • En la línea 1, se importa el módulo pymysql.

  • En la línea 3, se define en la constante USER, del usuario de conexión a la base de datos.

  • En la línea 4, se define en la constante PASSW, de la contraseña del usuario de conexión a la base de datos.

  • En la línea 5, se define en la constante HOST, la IP o dirección DNS de conexión al servidor de la base de datos.

  • En la línea 6, se define en la constante PORT, el puerto de conexión al servidor de la base de datos.

  • En la línea 7, se define en la constante DB, el nombre de la base de datos a cual conectar.

  • En la línea 8, se define en el método connect, el cual establece la conexión a la base de datos.

De esta forma se crea una cadena de conexión para MySQL para ser usada por el método connect.

6.2.3. Insertar registros

Si requiere insertar registro en una tabla, a continuación tiene un ejemplo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
"""Programa para la inserción de registro(s) de la tabla"""

import logging
import pymysql

logging.basicConfig(level=logging.INFO)

# Script CREATE TABLE SQL para crear tabla clientes
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS clientes (
    id INT AUTO_INCREMENT PRIMARY KEY,
    nombre VARCHAR(25) NOT NULL,
    apellido VARCHAR(25) NOT NULL,
    codigo_postal INT NOT NULL,
    telefono VARCHAR(20) NOT NULL
);"""

# Creando una lista de filas a ingresar
MULTIPLE_COLUMNS = [
    (1, "Leonardo", "Caballero", "5001", "+58-412-4734567"),
    (2, "Ana", "Poleo", "6302", "+58-426-5831297"),
    (3, "Manuel", "Matos", "4001", "+58-414-2360943"),
]

# Script INSERT SQL a usar al ingresar datos
INSERT_SQL_SCRIPTS = """INSERT INTO clientes VALUES (%s, %s, %s, %s, %s);"""


def insertar_registro(create_table_sql, insert_sql, insert_values):
    """Función para la inserción de registro de la tabla"""

    conexion = None
    credenciales = {
        "host": "localhost",
        "user": "root",
        "password": "root",
        "database": "sistema",
    }
    try:
        # Establecer la conexión con la base de datos
        conexion = pymysql.connect(
            host=credenciales["host"],
            user=credenciales["user"],
            password=credenciales["password"],
            database=credenciales["database"],
        )
        # Crear un objeto cursor para la base de datos
        cursor = conexion.cursor()
        logging.info(f"✅ ¡Conectado a la base de datos '{credenciales['database']}'!\n")
        # Crear la tabla productos si no existe
        cursor.execute(create_table_sql)
        # Confirmar la creación de la tabla
        conexion.commit()
        logging.info(
            f"✅ ¡Fue creo una tabla correctamente en la base de datos '{credenciales['database']}'!\n"
        )
        # Insertar nuevos registros en la tabla
        cursor.executemany(insert_sql, insert_values)
        # Confirmar la inserción de los registros
        conexion.commit()
        logging.info(
            f"✅ ¡Fueron insertado(s) {cursor.rowcount} registro(s) correctamente en la tabla!\n"
        )
        # Insertar un nuevo registro en la tabla
        cursor.execute(
            insert_sql, (4, "Liliana", "Andradez", "3105", "+58-414-6782473")
        )
        # Confirmar la inserción del registro
        conexion.commit()
        logging.info(
            f"✅ ¡Fueron insertado(s) {cursor.rowcount} registro(s) correctamente en la tabla!\n"
        )
        # Cerrar el cursor
        cursor.close()
    except pymysql.err.Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la inserción de registro(s) en la tabla!: {error}"
        )
    finally:
        if conexion:
            # Cerrar la conexión a la base de datos
            conexion.close()
            logging.info(
                f"✅ ¡La conexión MySQL a la base de datos '{credenciales['database']}' fue cerrada!"
            )


if __name__ == "__main__":
    insertar_registro(CREATE_TABLE_SQL, INSERT_SQL_SCRIPTS, MULTIPLE_COLUMNS)

Importante

Usted puede descargar el código usado en esta sección haciendo clic en el siguiente enlace:

Truco

Para ejecutar el código mysql_record_insert.py abra una consola de comando, acceda al directorio donde se encuentra el programa:

proyectos/
└── mysql/
    └── crud/
        └── mysql_record_insert.py

Si tiene la estructura de archivo previa, entonces ejecute el siguiente comando:

python3 mysql_record_insert.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Conectado a la base de datos 'sistema'!

INFO:root:✅ ¡Fue creo una tabla correctamente en la base de datos 'sistema'!

INFO:root:✅ ¡Fueron insertado(s) 3 registro(s) correctamente en la tabla!

INFO:root:✅ ¡Fueron insertado(s) 1 registro(s) correctamente en la tabla!

INFO:root:✅ ¡La conexión MySQL a la base de datos 'sistema' fue cerrada!

Puede probar si la base de datos sistema fue creada correctamente, ejecutando el siguiente comando correspondiente a tu sistema operativo:

docker exec -i mysql mysql -u root -proot -e "SHOW DATABASES;" mysql

Puede probar si el usuario root de la base de datos sistema fue creada correctamente, ejecutando el siguiente comando correspondiente a tu sistema operativo:

docker exec -i mysql mysql -u root -proot -e "SELECT user FROM mysql.user;" mysql

Puede probar si la tabla clientes en la base de datos sistema fue creada correctamente, ademas si sus registros fueron cargados en la tabla, ejecutando el siguiente comando correspondiente a tu sistema operativo:

docker exec -i mysql mysql -u root -proot -e "USE sistema; SHOW TABLES; SELECT * FROM clientes;" mysql

De esta forma puede ingresar registros en una tabla dentro una base de datos MySQL.

6.2.4. Consultar registros

Si requiere consultar registros de tabla, a continuación tiene un ejemplo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
"""Programa para la consulta de registro(s) de la tabla"""

import logging
import pymysql

logging.basicConfig(level=logging.INFO)

# Script SELECT SQL a usar al consultar datos
SELECT_SQL_SCRIPTS = """SELECT * FROM clientes;"""


def consultar_registro(select_sql):
    """Función para la consulta de registro(s) de la tabla"""

    conexion = None
    credenciales = {
        "host": "localhost",
        "user": "root",
        "password": "root",
        "database": "sistema",
    }
    try:
        # Establecer la conexión con la base de datos
        conexion = pymysql.connect(
            host=credenciales["host"],
            user=credenciales["user"],
            password=credenciales["password"],
            database=credenciales["database"],
        )
        # Crear un objeto cursor para ejecutar las consultas
        cursor = conexion.cursor()
        logging.info(f"✅ ¡Conectado a la base de datos '{credenciales['database']}'!\n")
        # Realizar consulta la tabla clientes
        cursor.execute(select_sql)
        # Recuperar los registros de la consulta
        registros = cursor.fetchall()
        # Mostrar los registros de la tabla
        print(f"📜 Total de filas son: {len(registros)} \n")
        print("📜 Mostrar cada fila: \n")
        for fila in registros:
            print(f"\tId: {fila[0]}")
            print(f"\tNombre: {fila[1]} {fila[2]}")
            print(f"\tCódigo postal: {fila[3]}")
            print(f"\tTeléfono: {fila[4]}\n")
        # Cerrar el cursor
        cursor.close()
    except pymysql.err.Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la consulta de registro(s) en la tabla!: {error}"
        )
    finally:
        if conexion:
            # Cerrar la conexión a la base de datos
            conexion.close()
            logging.info(
                f"✅ ¡La conexión MySQL a la base de datos '{credenciales['database']}' fue cerrada!"
            )


if __name__ == "__main__":
    consultar_registro(SELECT_SQL_SCRIPTS)

Importante

Usted puede descargar el código usado en esta sección haciendo clic en el siguiente enlace:

Truco

Para ejecutar el código mysql_record_select.py abra una consola de comando, acceda al directorio donde se encuentra el programa:

proyectos/
└── mysql/
    └── crud/
        └── mysql_record_select.py

Si tiene la estructura de archivo previa, entonces ejecute el siguiente comando:

python3 mysql_record_select.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Conectado a la base de datos 'sistema'!

📜 Total de filas son: 4

📜 Mostrar cada fila:

        Id: 1
        Nombre: Leonardo Caballero
        Código postal: 5001
        Teléfono: +58-412-4734567

        Id: 2
        Nombre: Ana Poleo
        Código postal: 6302
        Teléfono: +58-426-5831297

        Id: 3
        Nombre: Manuel Matos
        Código postal: 4001
        Teléfono: +58-414-2360943

        Id: 4
        Nombre: Liliana Andradez
        Código postal: 3105
        Teléfono: +58-414-6782473

INFO:root:✅ ¡La conexión MySQL a la base de datos 'sistema' fue cerrada!

De esta forma puede consultar registros en una tabla dentro una base de datos MySQL.

6.2.5. Actualizar registros

Si requiere actualizar registro de tabla, a continuación tiene un ejemplo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
"""Programa para la actualización de registro de la tabla"""

import logging
import pymysql

logging.basicConfig(level=logging.INFO)

# Creando una lista de filas a actualizar
MULTIPLE_COLUMNS = [
    ("5051", 1),
    ("6303", 2),
]

# Script UPDATE SQL a usar al actualizar datos
UPDATE_SQL_SCRIPTS = """UPDATE clientes SET codigo_postal = %s WHERE id = %s;"""


def actualizar_registro(update_sql, update_values):
    """Función para la actualización de registro de la tabla"""

    conexion = None
    credenciales = {
        "host": "localhost",
        "user": "root",
        "password": "root",
        "database": "sistema",
    }
    try:
        # Establecer la conexión con la base de datos
        conexion = pymysql.connect(
            host=credenciales["host"],
            user=credenciales["user"],
            password=credenciales["password"],
            database=credenciales["database"],
        )
        # Crear un cursor para la base de datos
        cursor = conexion.cursor()
        logging.info(f"✅ ¡Conectado a la base de datos '{credenciales['database']}'!\n")
        # Actualizar nuevos registros en la tabla
        cursor.executemany(update_sql, update_values)
        # Guardar los cambios en la base de datos
        conexion.commit()
        logging.info(
            f"✅ ¡Fueron actualizado(s) {cursor.rowcount} registro(s) correctamente en la tabla!\n"
        )
        # Cerrar el cursor
        cursor.close()
    except pymysql.err.Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la actualización de registro(s) en la tabla!: {error}"
        )
    finally:
        if conexion:
            # Cerrar la conexión a la base de datos
            conexion.close()
            logging.info(
                f"✅ ¡La conexión MySQL a la base de datos '{credenciales['database']}' fue cerrada!"
            )


if __name__ == "__main__":
    actualizar_registro(UPDATE_SQL_SCRIPTS, MULTIPLE_COLUMNS)

Importante

Usted puede descargar el código usado en esta sección haciendo clic en el siguiente enlace:

Truco

Para ejecutar el código mysql_record_update.py abra una consola de comando, acceda al directorio donde se encuentra el programa:

proyectos/
└── mysql/
    └── crud/
        └── mysql_record_update.py

Si tiene la estructura de archivo previa, entonces ejecute el siguiente comando:

python3 mysql_record_update.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Conectado a la base de datos 'sistema'!

INFO:root:✅ ¡Fueron actualizado(s) 2 registro(s) correctamente en la tabla!

INFO:root:✅ ¡La conexión MySQL a la base de datos 'sistema' fue cerrada!

De esta forma puede actualizar registros en una tabla dentro una base de datos MySQL.

6.2.6. Eliminar registros

Si requiere eliminar registro de tabla, a continuación tiene un ejemplo:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
"""Programa para la eliminación de registro de la tabla"""

import logging
import pymysql

logging.basicConfig(level=logging.INFO)

# Script DELETE SQL a usar al eliminar datos
DELETE_SQL_SCRIPTS = """DELETE FROM clientes WHERE id = 3;"""


def eliminar_registro(delete_sql):
    """Función para la eliminación de registro de la tabla"""

    conexion = None
    credenciales = {
        "host": "localhost",
        "user": "root",
        "password": "root",
        "database": "sistema",
    }
    try:
        # Establecer la conexión con la base de datos
        conexion = pymysql.connect(
            host=credenciales["host"],
            user=credenciales["user"],
            password=credenciales["password"],
            database=credenciales["database"],
        )
        # Crear un objeto cursor para ejecutar las eliminaciones
        cursor = conexion.cursor()
        logging.info(f"✅ ¡Conectado a la base de datos '{credenciales['database']}'!\n")
        # Eliminar un fila de registro simple
        cursor.execute(delete_sql)
        # Guardar los cambios en la base de datos
        conexion.commit()
        logging.info("✅ ¡Registro eliminado correctamente!\n")
        # Cerrar el cursor
        cursor.close()
    except pymysql.err.Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la eliminación de registro(s) en la tabla!: {error}"
        )
    finally:
        if conexion:
            # Cerrar la conexión a la base de datos
            conexion.close()
            logging.info(
                f"✅ ¡La conexión MySQL a la base de datos '{credenciales['database']}' fue cerrada!"
            )


if __name__ == "__main__":
    eliminar_registro(DELETE_SQL_SCRIPTS)

Importante

Usted puede descargar el código usado en esta sección haciendo clic en el siguiente enlace:

Truco

Para ejecutar el código mysql_record_delete.py abra una consola de comando, acceda al directorio donde se encuentra el programa:

proyectos/
└── mysql/
    └── crud/
        └── mysql_record_delete.py

Si tiene la estructura de archivo previa, entonces ejecute el siguiente comando:

python3 mysql_record_delete.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Conectado a la base de datos 'sistema'!

INFO:root:✅ ¡Registro eliminado correctamente!

INFO:root:✅ ¡La conexión MySQL a la base de datos 'sistema' fue cerrada!

De esta forma puede eliminar registros en una tabla dentro una base de datos MySQL.

Nota

Así de esta forma puede realizar las operaciones de ingresar, consultar, actualizar y eliminar registro en una tabla en una base de datos MySQL de forma separada en programas Python, en la siguiente práctica se mostrara un caso real de uso de todos estas operaciones en un solo programa Python.

6.2.7. Práctica - Caso real

A continuación se presenta una práctica más real de implementar el uso de proyectos con MySQL en Python:

6.2.7.1. Estructura de archivos

Para crear la estructura de archivos del proyecto MySQL debe ejecutar los siguientes comandos:

Crear y acceder al directorio sistema en un solo comando, ejecutando el siguiente comando:

mkdir -p ~/proyectos/mysql/sistema && cd $_

El comando anterior crea la siguiente estructura de directorios:

proyectos/
└── mysql/
    └── sistema/

Si tiene la estructura de archivo previa, entonces puede continuar con la siguiente sección.

A continuación se presenta y explica el uso de cada archivo para este proyecto:

Archivo .env.example

Archivo plantilla dotenv, es un archivo de configuración de variables de entorno para el proyecto. Además, es usado para establecer variables de entorno con Docker, Docker Compose y del paquete adicional python-dotenv.

1
2
3
4
5
6
7
8
### MySQL #############################
MYSQL_VERSION=latest
MYSQL_CONTAINER=mysql
USER=root
PASSW=root
HOST=localhost
PORT=3306
DB=sistema

Archivo requirements.txt

Archivo de requirements.txt de la herramienta de gestión de paquetes pip.

1
2
3
4
# Install the packages from PyPi repository.
python-dotenv==1.0.1
PyMySQL==1.1.1
cryptography==44.0.1

Archivo settings.py

Módulo de configuraciones del programa.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
import os
from dotenv import load_dotenv

load_dotenv()

# Usuario de conexión a la base de datos.
USER = os.getenv("USER")

# Contraseña del usuario de conexión a la base de datos.
PASSW = os.getenv("PASSW")

# IP o dirección DNS de conexión al servidor de la base de datos.
HOST = os.getenv("HOST")

# Puerto de conexión al servidor de la base de datos, por defecto es '3306'.
PORT = int(os.getenv("PORT", 3306))

# Nombre de la base de datos a cual conectar.
DB = os.getenv("DB")

# Script CREATE DATABASE SQL para crear la base de datos
CREATE_DATABASE_SQL = """CREATE DATABASE %s;"""

# Script CREATE TABLE SQL para crear tabla(s)
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS clientes (
    id INT AUTO_INCREMENT PRIMARY KEY,
    nombre VARCHAR(25) NOT NULL,
    apellido VARCHAR(25) NOT NULL,
    codigo_postal INT NOT NULL,
    telefono VARCHAR(20) NOT NULL
);"""

# Lista de filas a ingresar
INSERT_MULTIPLE_COLUMNS = [
    (1, "Leonardo", "Caballero", "5001", "+58-412-4734567"),
    (2, "Ana", "Poleo", "6302", "+58-426-5831297"),
    (3, "Manuel", "Matos", "4001", "+58-414-2360943"),
]

# Script INSERT SQL a usar al ingresar datos
INSERT_SQL_SCRIPTS = """INSERT INTO clientes VALUES (%s, %s, %s, %s, %s);"""

# Script SELECT SQL a usar al consultar datos
SELECT_SQL_SCRIPTS = """SELECT * FROM clientes;"""

# Lista de filas a actualizar
UPDATE_MULTIPLE_COLUMNS = [
    ("5051", 1),
    ("6303", 2),
]

# Script UPDATE SQL a usar al actualizar datos
UPDATE_SQL_SCRIPTS = """UPDATE clientes SET codigo_postal = %s WHERE id = %s;"""

# Script DELETE SQL a usar al eliminar datos
DELETE_SQL_SCRIPTS = """DELETE FROM clientes WHERE id = 3;"""

Archivo main.py

Módulo principal del programa.

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
"""Programa para realizar operaciones a base de datos MySQL"""

import logging
from settings import (
    USER,
    PASSW,
    HOST,
    PORT,
    DB,
    CREATE_DATABASE_SQL,
    CREATE_TABLE_SQL,
    INSERT_MULTIPLE_COLUMNS,
    INSERT_SQL_SCRIPTS,
    SELECT_SQL_SCRIPTS,
    UPDATE_MULTIPLE_COLUMNS,
    UPDATE_SQL_SCRIPTS,
    DELETE_SQL_SCRIPTS,
)
from pymysql.constants.ER import DBACCESS_DENIED_ERROR, BAD_DB_ERROR
from pymysql import (
    Error,
    OperationalError,
    ProgrammingError,
    IntegrityError,
    connect,
    err,
)

logging.basicConfig(level=logging.INFO)


def crear_conexion(servidor, puerto, usuario, contrasena, bd):
    """Crear conexión con un servidor MySQL

    Args:
        servidor (str): IP o dirección DNS de conexión al servidor de la base de datos.
        puerto (int): Puerto de conexión al servidor de la base de datos.
        usuario (str): Usuario de conexión a la base de datos.
        contrasena (str): Contraseña del usuario de conexión a la base de datos.
        bd (str): Nombre de la base de datos a cual conectar.

    Returns:
        conexion_bd (Connection): Representación de un socket con un servidor MySQL
    """
    conexion_bd = None
    credenciales = {
        "host": servidor,
        "port": puerto,
        "user": usuario,
        "password": contrasena,
        "database": bd,
    }
    try:
        # Establecer la conexión con la base de datos
        conexion_bd = connect(
            host=credenciales["host"],
            port=credenciales["port"],
            user=credenciales["user"],
            password=credenciales["password"],
            database=credenciales["database"],
        )
        logging.info(
            f"✅ ¡Conexión a la base de datos '{credenciales['database']}' fue exitosa!\n"
        )
    except Error as err:
        if err.args[0] == DBACCESS_DENIED_ERROR:
            logging.error(
                "\x1b[1;31mERROR: ¡Algo está mal con su nombre de usuario o contraseña!"
            )
        elif err.args[0] == BAD_DB_ERROR:
            logging.error("\x1b[1;31m❌ ERROR: ¡La base de datos no existe!")
        else:
            logging.error(f"\x1b[1;31m❌ ERROR: Se produjo lo siguiente: '{err}'")
    return conexion_bd


def crear_base_datos(conexion_bd, create_database_sql, bd):
    """Creación la base de datos

    Args:
        conexion_bd (Connection): Representación de un socket con un servidor MySQL
        create_database_sql (str): Script CREATE DATABASE SQL para crear la base de datos
        bd (str): Nombre de la base de datos a crear.
    """
    conexion_bd.autocommit = True
    # Crear un objeto cursor para ejecutar script SQL
    cursor = conexion_bd.cursor()
    try:
        # Crear una base de datos en el servidor MySQL
        cursor.execute(create_database_sql, bd)
        logging.info(f"✅ ¡Creación exitosa de la base de datos '{bd}'!\n")
    except ProgrammingError as e:
        logging.error(f"❌ ERROR: ¡Se produjo una falla de programación: '{e}'!")
    except OperationalError as e:
        logging.error(f"❌ ERROR: Se produjo lo siguiente: '{e}'")
    return conexion_bd


def crear_tablas(conexion_bd, create_table_sql):
    """Creación de tabla(s) dentro de la base de datos

    Args:
        conexion_bd (Connection): Representación conexión a la base de datos MySQL
        create_table_sql (str): Script CREATE TABLE SQL para crear tabla(s)
    """
    try:
        # Crear un objeto cursor para ejecutar script SQL
        cursor = conexion_bd.cursor()
        # Crear la tabla(s) si no existe
        cursor.execute(create_table_sql)
        # Hacer persistentes los cambios en la base de datos
        conexion_bd.commit()
        logging.info(
            f"✅ ¡Fueron creado(s) {cursor.rowcount} tabla(s) correctamente en la base de datos!\n"
        )
        # Cerrar el cursor
        cursor.close()
    except Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la creación de tabla(s) en la base de datos!: {error}"
        )


def insertar_registro(conexion_bd, insert_values, insert_sql):
    """Función para la inserción de registro de la tabla

    Args:
        conexion_bd (Connection): Representación conexión a la base de datos MySQL
        insert_values (list): Lista de filas a ingresar
        insert_sql (str): Script INSERT SQL a usar al ingresar datos
    """
    try:
        # Crear un objeto cursor para ejecutar script SQL
        cursor = conexion_bd.cursor()
        # Insertar nuevos registros en la tabla
        cursor.executemany(insert_sql, insert_values)
        logging.info(
            f"✅ ¡Fueron insertado(s) {cursor.rowcount} registro(s) correctamente en la tabla!\n"
        )
        # Insertar un nuevo registro en la tabla
        cursor.execute(
            insert_sql, (4, "Liliana", "Andradez", "3105", "+58-414-6782473")
        )
        # Hacer persistentes los cambios en la base de datos
        conexion_bd.commit()
        logging.info(
            f"✅ ¡Fueron insertado(s) {cursor.rowcount} registro(s) correctamente en la tabla!"
        )
        # Cerrar el cursor
        cursor.close()
    except IntegrityError as error:
        logging.error(f"❌ ERROR: ¡Registro duplicado por clave primaria!: {error}")
    except Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la inserción de registro(s) en la tabla!: {error}"
        )


def consultar_registro(conexion_bd, select_sql):
    """Función para la consulta de registro(s) de la tabla

    Args:
        conexion_bd (Connection): Representación conexión a la base de datos MySQL
        select_sql (str): Script SELECT SQL a usar al consultar datos
    """
    try:
        # Crear un objeto cursor para ejecutar script SQL
        cursor = conexion_bd.cursor()
        # Realizar consulta la tabla clientes
        cursor.execute(select_sql)
        # Recuperar los registros de la consulta
        registros = cursor.fetchall()
        # Mostrar los registros de la tabla
        print(f"\n📜 Total de filas son: {len(registros)} \n")
        print("📜 Mostrar cada fila: \n")
        for fila in registros:
            print(f"\tId: {fila[0]}")
            print(f"\tNombre: {fila[1]} {fila[2]}")
            print(f"\tCódigo postal: {fila[3]}")
            print(f"\tTeléfono: {fila[4]}\n")
        # Cerrar el cursor
        cursor.close()
    except Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la consulta de registro(s) en la tabla!: {error}"
        )


def actualizar_registro(conexion_bd, update_values, update_sql):
    """Función para la actualización de registro de la tabla

    Args:
        conexion_bd (Connection): Representación conexión a la base de datos MySQL
        update_values (list): Lista de filas a actualizar
        update_sql (str): Script UPDATE SQL a usar al actualizar datos
    """
    try:
        # Crear un objeto cursor para ejecutar script SQL
        cursor = conexion_bd.cursor()
        # Actualizar nuevos registros en la tabla
        cursor.executemany(update_sql, update_values)
        # Hacer persistentes los cambios en la base de datos
        conexion_bd.commit()
        logging.info(
            f"✅ ¡Fueron actualizado(s) {cursor.rowcount} registro(s) correctamente en la tabla!\n"
        )
        # Cerrar el cursor
        cursor.close()
    except Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la actualización de registro(s) en la tabla!: {error}"
        )


def eliminar_registro(conexion_bd, delete_sql):
    """Función para la eliminación de registro de la tabla

    Args:
        conexion_bd (Connection): Representación conexión a la base de datos MySQL
        delete_sql (str): Script DELETE SQL a usar al eliminar datos
    """
    try:
        # Crear un objeto cursor para ejecutar script SQL
        cursor = conexion_bd.cursor()
        # Eliminar un fila de registro simple
        cursor.execute(delete_sql)
        # Hacer persistentes los cambios en la base de datos
        conexion_bd.commit()
        logging.info("✅ ¡Registro eliminado correctamente!\n")
        # Cerrar el cursor
        cursor.close()
    except Error as error:
        logging.error(
            f"❌ ERROR: ¡Fallo la eliminación de registro(s) en la tabla!: {error}"
        )


if __name__ == "__main__":
    conexion = None
    try:
        # Crear conexión al servidor MySQL
        conexion = crear_conexion(HOST, PORT, USER, PASSW, DB)
        # Crear la base de datos
        # crear_base_datos(conexion, CREATE_DATABASE_SQL, DB)
        # Crear la tabla dentro de la base de datos
        crear_tablas(conexion, CREATE_TABLE_SQL)
        insertar_registro(conexion, INSERT_MULTIPLE_COLUMNS, INSERT_SQL_SCRIPTS)
        consultar_registro(conexion, SELECT_SQL_SCRIPTS)
        actualizar_registro(conexion, UPDATE_MULTIPLE_COLUMNS, UPDATE_SQL_SCRIPTS)
        eliminar_registro(conexion, DELETE_SQL_SCRIPTS)
    except Error as e:
        logging.error(
            f"❌ ERROR: ¡Se produjo un falla al establecer la conexión a la base de datos '{DB}': '{e}'!"
        )
    finally:
        if conexion:
            # Cerrar la conexión a la base de datos
            conexion.close()
            logging.info(
                f"✅ ¡La conexión PostgreSQL a la base de datos '{DB}' fue cerrada!"
            )

Archivo docker-compose.yml

Para instalar el servidor MySQL existen varias formas de realizarlo, para en este caso se realizara con la tecnología Docker. Esto significa que debe instalar en tu sistema operativo:

El primer paso para configurar un entorno de desarrollo con Docker Compose es crear el archivo de configuración docker-compose.yml. Este archivo define los servicios, contenedores, redes y volúmenes necesarios para tu aplicación.

A continuación se presenta el archivo docker-compose.yml con la configuración necesaria:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
services:
  db:
    env_file:
      - .env
    image: "mysql:${MYSQL_VERSION}"
    container_name: ${MYSQL_CONTAINER}
    restart: always
    environment:
      - MYSQL_ROOT_PASSWORD=${PASSW}
      - MYSQL_DATABASE=${DB}
      - MYSQL_USER=${USER}
      - MYSQL_PASSWORD=${PASSW}
    ports:
      - "${PORT}:3306"
    volumes:
      - my_data:/var/lib/mysql
    healthcheck:
      test: mysqladmin ping -h ${HOST} -u $${USER} --password=$${PASSW}
      start_period: 5s
      interval: 5s
      timeout: 5s
      retries: 55

volumes:
  my_data:

Para ejecutar el código del proyecto llamado sistema abra una consola de comando, cree la siguiente estructura de directorio y acceda al mismo donde se encuentra el programa:

proyectos/
└── mysql/
    └── sistema/
        ├── docker-compose.yml
        ├── __init__.py
        ├── .env.example
        ├── main.py
        ├── requirements.txt
        └── settings.py

Si tiene la estructura de archivo previa, entonces puede continuar los procesos de instalación, configuración y ejecución del código fuente.

Antes de ejecutar debes instalar sus dependencias, con el siguiente comando:

pip3 install -r requirements.txt

Además debe crear el archivo .env en base a la plantilla .env.example` y editarlo, con el siguiente comando:

cp .env.example .env && nano .env

Truco

El archivo .env se definen las configuraciones de conexión a la base de datos, puede modificarlo cambiar valores de la conexión.

Debe crear y editar el archivo docker-compose.yml, con el siguiente comando:

Truco

Para ejecutar el comando del instalador del servidor MySQL con Docker debe crear un archivo llamado docker-compose.yml en el directorio sistema con el contenido anterior de dicho archivo, ejecutando el siguiente comando:

nano docker-compose.yml

Truco

Si tiene creado el archivo con el contenido, entonces puede ejecutar la instalación un servidor MySQL, ejecutando el siguiente comando:

docker-compose up -d

De esta forma crea el contenedor Docker llamado mysql, necesario para ejecutar el script Python.

Truco

Para ejecutar el código fuente de esta práctica debe invocar al módulo main.py, abra una consola de comando, acceda al directorio donde se encuentra la estructura previa y ejecute el siguiente comando:

python3 main.py

El anterior código al ejecutar debe mostrar el siguiente mensaje:

INFO:root:✅ ¡Conexión a la base de datos 'sistema' fue exitosa!

INFO:root:✅ ¡Fueron creado(s) 1 tabla(s) correctamente en la base de datos!

INFO:root:✅ ¡Fueron insertado(s) 3 registro(s) correctamente en la tabla!

INFO:root:✅ ¡Fueron insertado(s) 1 registro(s) correctamente en la tabla!

📜 Total de filas son: 4

📜 Mostrar cada fila:

        Id: 1
        Nombre: Leonardo Caballero
        Código postal: 5001
        Teléfono: +58-412-4734567

        Id: 2
        Nombre: Ana Poleo
        Código postal: 6302
        Teléfono: +58-426-5831297

        Id: 3
        Nombre: Manuel Matos
        Código postal: 4001
        Teléfono: +58-414-2360943

        Id: 4
        Nombre: Liliana Andradez
        Código postal: 3105
        Teléfono: +58-414-6782473

INFO:root:✅ ¡Fueron actualizado(s) 2 registro(s) correctamente en la tabla!

INFO:root:✅ ¡Registro eliminado correctamente!

INFO:root:✅ ¡La conexión MySQL a la base de datos 'sistema' fue cerrada!

Así de esta forma puede ingresar, consultar, actualizar y eliminar registro en una tabla usando MySQL.

Importante

Usted puede descargar el código usado en esta sección haciendo clic en los siguientes enlaces:

Así de esta forma puede replicar una práctica real de un proyecto para realizar operaciones en una base de datos MySQL, aplicando buenas prácticas de código funcional.

Ver también

Consulte la sección de lecturas suplementarias del entrenamiento para ampliar su conocimiento en esta temática.

¿Cómo puedo ayudar?

¡Mi soporte está aquí para ayudar!

Mi horario de oficina es de lunes a sábado, de 9 AM a 5 PM. UTM - Madrid, España.

La hora aquí es actualmente 7:35 PM UTM.

Mi objetivo es responder a todos los mensajes dentro de un día hábil.

Contáctenos en la sección de soporte
Contrata mi increíble soporte profesional