Python

Tutorial Python: Acceso a MySQL con MySQL Connector

Aprende a conectar Python con MySQL usando MySQL Connector. Configura conexiones, realiza operaciones CRUD y gestiona transacciones de forma segura y eficiente.

Configuración de la conexión

La conexión a una base de datos MySQL desde Python es el primer paso fundamental para cualquier aplicación que necesite persistencia de datos. MySQL Connector/Python es la biblioteca oficial desarrollada por Oracle que permite establecer conexiones entre aplicaciones Python y servidores MySQL.

Para comenzar a trabajar con MySQL desde Python, primero necesitamos configurar correctamente la conexión. Esto implica instalar el conector adecuado, establecer los parámetros de conexión y gestionar la conexión de manera eficiente.

Instalación de MySQL Connector/Python

Antes de configurar la conexión, debemos instalar el paquete necesario. Podemos hacerlo fácilmente utilizando pip:

pip install mysql-connector-python

Este comando instalará la versión más reciente del conector oficial. Es importante usar el conector oficial en lugar de alternativas de terceros para garantizar la compatibilidad y el soporte a largo plazo.

Importación del módulo

Una vez instalado, podemos importar el módulo en nuestro código Python:

import mysql.connector
from mysql.connector import Error

La segunda línea importa la clase Error, que nos permitirá manejar excepciones específicas de MySQL Connector.

Estableciendo una conexión básica

La forma más simple de establecer una conexión es proporcionando los parámetros básicos de conexión:

try:
    connection = mysql.connector.connect(
        host="localhost",      # Servidor MySQL (dirección IP o nombre de dominio)
        user="usuario",        # Nombre de usuario
        password="contraseña", # Contraseña
        database="mi_base"     # Nombre de la base de datos
    )
    
    if connection.is_connected():
        db_info = connection.get_server_info()
        print(f"Conectado a MySQL versión {db_info}")
        
        # Crear un cursor para ejecutar consultas
        cursor = connection.cursor()
        
        # Ejemplo: obtener la versión de MySQL
        cursor.execute("SELECT VERSION()")
        version = cursor.fetchone()
        print(f"Versión de la base de datos: {version[0]}")
        
except Error as e:
    print(f"Error al conectar a MySQL: {e}")
    
finally:
    # Cerrar la conexión y el cursor si existen
    if 'connection' in locals() and connection.is_connected():
        cursor.close()
        connection.close()
        print("Conexión a MySQL cerrada")

Este ejemplo muestra un patrón completo de conexión que incluye:

Establecimiento de la conexión con parámetros básicos
Verificación de que la conexión se estableció correctamente
Creación de un cursor para ejecutar consultas
Manejo de excepciones
Cierre adecuado de recursos en un bloque finally

Parámetros de conexión avanzados

MySQL Connector/Python ofrece parámetros adicionales para personalizar el comportamiento de la conexión:

connection = mysql.connector.connect(
    host="localhost",
    user="usuario",
    password="contraseña",
    database="mi_base",
    port=3306,                # Puerto personalizado (3306 es el predeterminado)
    auth_plugin="mysql_native_password",  # Método de autenticación
    charset="utf8mb4",        # Conjunto de caracteres
    collation="utf8mb4_unicode_ci",  # Colación
    autocommit=False,         # Desactivar autocommit
    pool_name="mypool",       # Nombre del pool de conexiones
    pool_size=5,              # Tamaño del pool (número de conexiones)
    buffered=True,            # Usar cursores almacenados en búfer
    raw=False,                # Convertir valores a tipos Python
    use_pure=True,            # Usar implementación pura de Python
    connection_timeout=10,    # Tiempo de espera para la conexión en segundos
    ssl_ca="/path/to/ca.pem", # Certificado CA para conexiones SSL
    ssl_cert="/path/to/cert.pem",  # Certificado cliente
    ssl_key="/path/to/key.pem"     # Clave privada
)

No es necesario especificar todos estos parámetros; solo debes incluir los que necesites para tu caso específico.

Uso de un diccionario de configuración

Una práctica recomendada es mantener los parámetros de conexión en un diccionario separado, lo que facilita la gestión y modificación:

config = {
    'user': 'usuario',
    'password': 'contraseña',
    'host': 'localhost',
    'database': 'mi_base',
    'raise_on_warnings': True
}

try:
    connection = mysql.connector.connect(**config)
    # Resto del código...
except Error as e:
    print(f"Error: {e}")

Este enfoque es especialmente útil cuando necesitas cargar la configuración desde archivos externos o variables de entorno.

Conexión usando un archivo de configuración

Para aplicaciones más grandes, es recomendable almacenar la configuración en un archivo separado:

# config.py
mysql_config = {
    'user': 'usuario',
    'password': 'contraseña',
    'host': 'localhost',
    'database': 'mi_base'
}

Luego, en tu script principal:

from config import mysql_config
import mysql.connector

try:
    connection = mysql.connector.connect(**mysql_config)
    # Resto del código...
except mysql.connector.Error as err:
    print(f"Error: {err}")

Conexión con manejo de contexto (context manager)

Python 3 permite utilizar el patrón de contexto (with) para gestionar recursos automáticamente. MySQL Connector/Python soporta este patrón desde la versión 8.0:

import mysql.connector

config = {
    'user': 'usuario',
    'password': 'contraseña',
    'host': 'localhost',
    'database': 'mi_base'
}

try:
    with mysql.connector.connect(**config) as connection:
        with connection.cursor() as cursor:
            cursor.execute("SELECT * FROM usuarios LIMIT 5")
            for row in cursor:
                print(row)
        # No es necesario cerrar el cursor, se cierra automáticamente
    # No es necesario cerrar la conexión, se cierra automáticamente
except mysql.connector.Error as err:
    print(f"Error: {err}")

Este enfoque garantiza que los recursos se liberen correctamente, incluso si ocurre una excepción.

Conexión a través de un pool de conexiones

Para aplicaciones que requieren múltiples conexiones simultáneas, MySQL Connector/Python ofrece un pool de conexiones que mejora el rendimiento:

from mysql.connector import pooling

dbconfig = {
    "host": "localhost",
    "user": "usuario",
    "password": "contraseña",
    "database": "mi_base"
}

try:
    # Crear un pool con 5 conexiones
    connection_pool = pooling.MySQLConnectionPool(
        pool_name="pynative_pool",
        pool_size=5,
        **dbconfig
    )
    
    # Obtener una conexión del pool
    connection = connection_pool.get_connection()
    
    if connection.is_connected():
        print("Conexión obtenida del pool")
        
        cursor = connection.cursor()
        cursor.execute("SELECT * FROM usuarios LIMIT 3")
        records = cursor.fetchall()
        
        for record in records:
            print(record)
            
        # Devolver la conexión al pool (no cerrarla)
        cursor.close()
        connection.close()
        print("Conexión devuelta al pool")
        
except Error as e:
    print(f"Error al conectar con MySQL: {e}")

El uso de pools es especialmente útil en aplicaciones web o servicios que manejan múltiples solicitudes concurrentes.

Conexión segura con SSL/TLS

Para conexiones que requieren seguridad adicional, especialmente en entornos de producción o cuando la base de datos está en un servidor remoto, podemos configurar SSL/TLS:

ssl_config = {
    "user": "usuario",
    "password": "contraseña",
    "host": "db.ejemplo.com",
    "database": "mi_base",
    "client_flags": [mysql.connector.ClientFlag.SSL],
    "ssl_ca": "/path/to/ca.pem",
    "ssl_cert": "/path/to/client-cert.pem",
    "ssl_key": "/path/to/client-key.pem",
    "ssl_verify_cert": True
}

try:
    connection = mysql.connector.connect(**ssl_config)
    if connection.is_connected():
        print("Conectado de forma segura mediante SSL")
except Error as e:
    print(f"Error en la conexión SSL: {e}")

Verificación y diagnóstico de la conexión

Es importante verificar que la conexión se ha establecido correctamente y diagnosticar problemas:

try:
    connection = mysql.connector.connect(**config)
    
    if connection.is_connected():
        db_info = connection.get_server_info()
        print(f"Conectado a MySQL versión: {db_info}")
        
        cursor = connection.cursor()
        cursor.execute("SELECT DATABASE();")
        db_name = cursor.fetchone()[0]
        print(f"Base de datos actual: {db_name}")
        
        # Verificar privilegios del usuario
        cursor.execute("SHOW GRANTS FOR CURRENT_USER;")
        for grant in cursor:
            print(f"Privilegio: {grant[0]}")
            
except Error as e:
    print(f"Error al conectar: {e}")
    # Diagnóstico adicional
    if "Access denied" in str(e):
        print("Problema de autenticación: verifica usuario y contraseña")
    elif "Unknown database" in str(e):
        print("La base de datos especificada no existe")
    elif "Can't connect to MySQL server" in str(e):
        print("No se puede conectar al servidor: verifica host y puerto")

Este código proporciona información detallada sobre la conexión y ayuda a diagnosticar problemas comunes.

Manejo de timeouts y reconexión

En aplicaciones de larga duración, las conexiones pueden cerrarse por inactividad. Podemos implementar una lógica de reconexión:

def get_connection(config, max_attempts=3):
    attempt = 1
    while attempt <= max_attempts:
        try:
            connection = mysql.connector.connect(**config)
            if connection.is_connected():
                print(f"Conexión establecida (intento {attempt})")
                return connection
        except Error as e:
            print(f"Intento {attempt} fallido: {e}")
            attempt += 1
            if attempt <= max_attempts:
                print(f"Reintentando en 5 segundos...")
                import time
                time.sleep(5)
    
    raise Exception(f"No se pudo establecer conexión después de {max_attempts} intentos")

# Uso
try:
    conn = get_connection(config)
    # Usar la conexión...
except Exception as e:
    print(f"Error final: {e}")

Esta función intenta conectarse varias veces antes de rendirse, lo que es útil en entornos con conexiones inestables.

Operaciones CRUD básicas

Una vez establecida la conexión con MySQL, el siguiente paso es realizar operaciones CRUD (Create, Read, Update, Delete) que constituyen la base de cualquier interacción con bases de datos. Estas operaciones permiten manipular datos de manera estructurada y eficiente.

Preparación del entorno

Antes de comenzar con las operaciones CRUD, necesitamos tener una conexión establecida y un cursor para ejecutar las consultas:

import mysql.connector
from mysql.connector import Error

try:
    connection = mysql.connector.connect(
        host="localhost",
        user="usuario",
        password="contraseña",
        database="mi_base"
    )
    
    # Crear un cursor para ejecutar consultas
    cursor = connection.cursor()
    
    # Aquí irán nuestras operaciones CRUD
    
except Error as e:
    print(f"Error: {e}")
finally:
    if 'connection' in locals() and connection.is_connected():
        cursor.close()
        connection.close()

Operación Create (Insertar datos)

La operación Create permite añadir nuevos registros a una tabla. Existen varias formas de realizar inserciones en MySQL con Python:

Inserción de un solo registro

try:
    # Definir la consulta SQL con marcadores de posición
    query = "INSERT INTO empleados (nombre, apellido, email, departamento, salario) VALUES (%s, %s, %s, %s, %s)"
    
    # Datos a insertar
    valores = ("Ana", "García", "ana.garcia@ejemplo.com", "Marketing", 45000)
    
    # Ejecutar la consulta
    cursor.execute(query, valores)
    
    # Confirmar los cambios en la base de datos
    connection.commit()
    
    # Obtener el ID del último registro insertado
    ultimo_id = cursor.lastrowid
    print(f"Registro insertado con ID: {ultimo_id}")
    
except Error as e:
    print(f"Error al insertar: {e}")
    # Deshacer cambios en caso de error
    connection.rollback()

Es importante notar el uso de marcadores de posición (%s) en lugar de concatenar directamente los valores en la consulta. Esto previene ataques de inyección SQL y maneja automáticamente el escapado de caracteres especiales.

Inserción de múltiples registros

Para insertar varios registros de manera eficiente, podemos usar executemany():

try:
    query = "INSERT INTO empleados (nombre, apellido, email, departamento, salario) VALUES (%s, %s, %s, %s, %s)"
    
    # Lista de tuplas con los datos a insertar
    valores = [
        ("Carlos", "Rodríguez", "carlos.rodriguez@ejemplo.com", "Ventas", 38000),
        ("Laura", "Martínez", "laura.martinez@ejemplo.com", "Desarrollo", 52000),
        ("Miguel", "López", "miguel.lopez@ejemplo.com", "Finanzas", 48000)
    ]
    
    # Ejecutar la consulta múltiple
    cursor.executemany(query, valores)
    
    # Confirmar los cambios
    connection.commit()
    
    # Mostrar número de registros insertados
    print(f"Se insertaron {cursor.rowcount} registros")
    
except Error as e:
    print(f"Error al insertar múltiples registros: {e}")
    connection.rollback()

Este método es mucho más eficiente que realizar múltiples llamadas a execute(), ya que reduce el número de viajes entre la aplicación y el servidor de base de datos.

Operación Read (Leer datos)

La operación Read permite consultar información almacenada en la base de datos. Hay varias formas de recuperar datos:

Consulta básica (SELECT)

try:
    # Consulta simple
    cursor.execute("SELECT id, nombre, apellido, departamento FROM empleados")
    
    # Recuperar todos los resultados
    resultados = cursor.fetchall()
    
    # Mostrar los resultados
    print("ID | Nombre | Apellido | Departamento")
    print("-" * 40)
    for fila in resultados:
        print(f"{fila[0]} | {fila[1]} | {fila[2]} | {fila[3]}")
    
    # Mostrar el número total de registros
    print(f"\nTotal de registros: {len(resultados)}")
    
except Error as e:
    print(f"Error al consultar datos: {e}")

Métodos de recuperación de datos

MySQL Connector proporciona tres métodos principales para recuperar resultados:

# Recuperar todas las filas
cursor.execute("SELECT * FROM empleados")
todas_filas = cursor.fetchall()

# Recuperar solo la primera fila
cursor.execute("SELECT * FROM empleados")
primera_fila = cursor.fetchone()

# Recuperar un número específico de filas
cursor.execute("SELECT * FROM empleados")
algunas_filas = cursor.fetchmany(size=3)  # Recupera 3 filas

Consulta con filtros

Para filtrar resultados, usamos la cláusula WHERE con parámetros:

try:
    # Consulta con filtro
    departamento = "Desarrollo"
    salario_minimo = 40000
    
    query = "SELECT nombre, apellido, salario FROM empleados WHERE departamento = %s AND salario > %s"
    cursor.execute(query, (departamento, salario_minimo))
    
    resultados = cursor.fetchall()
    
    print(f"\nEmpleados del departamento {departamento} con salario > {salario_minimo}:")
    for empleado in resultados:
        print(f"{empleado[0]} {empleado[1]}: {empleado[2]}€")
    
except Error as e:
    print(f"Error en consulta filtrada: {e}")

Uso de cursores con nombre de columna

Por defecto, los resultados se devuelven como tuplas, lo que puede dificultar la lectura del código. Podemos usar un cursor que devuelva diccionarios:

try:
    # Crear un cursor que devuelva diccionarios
    cursor_dict = connection.cursor(dictionary=True)
    
    cursor_dict.execute("SELECT id, nombre, apellido, email FROM empleados LIMIT 3")
    resultados = cursor_dict.fetchall()
    
    for empleado in resultados:
        print(f"ID: {empleado['id']}")
        print(f"Nombre: {empleado['nombre']} {empleado['apellido']}")
        print(f"Email: {empleado['email']}")
        print("-" * 30)
    
    cursor_dict.close()
    
except Error as e:
    print(f"Error con cursor de diccionario: {e}")

Este tipo de cursor facilita el acceso a los datos por nombre de columna en lugar de por posición.

Operación Update (Actualizar datos)

La operación Update permite modificar registros existentes en la base de datos:

try:
    # Actualizar un solo registro
    query = "UPDATE empleados SET salario = %s, departamento = %s WHERE id = %s"
    
    # Nuevos valores y condición
    nuevo_salario = 55000
    nuevo_departamento = "Dirección"
    id_empleado = 5
    
    cursor.execute(query, (nuevo_salario, nuevo_departamento, id_empleado))
    
    # Confirmar cambios
    connection.commit()
    
    # Verificar si se actualizó algún registro
    if cursor.rowcount > 0:
        print(f"Registro con ID {id_empleado} actualizado correctamente")
    else:
        print(f"No se encontró ningún empleado con ID {id_empleado}")
    
except Error as e:
    print(f"Error al actualizar: {e}")
    connection.rollback()

Actualización de múltiples registros

También podemos actualizar varios registros que cumplan cierta condición:

try:
    # Actualizar múltiples registros
    query = "UPDATE empleados SET salario = salario * 1.05 WHERE departamento = %s"
    
    departamento = "Ventas"
    cursor.execute(query, (departamento,))
    
    connection.commit()
    
    print(f"Se actualizaron {cursor.rowcount} registros del departamento {departamento}")
    print("Se aplicó un aumento del 5% a sus salarios")
    
except Error as e:
    print(f"Error al actualizar múltiples registros: {e}")
    connection.rollback()

Operación Delete (Eliminar datos)

La operación Delete permite eliminar registros de una tabla:

try:
    # Eliminar un registro específico
    query = "DELETE FROM empleados WHERE id = %s"
    
    id_a_eliminar = 7
    cursor.execute(query, (id_a_eliminar,))
    
    connection.commit()
    
    if cursor.rowcount > 0:
        print(f"Empleado con ID {id_a_eliminar} eliminado correctamente")
    else:
        print(f"No se encontró ningún empleado con ID {id_a_eliminar}")
    
except Error as e:
    print(f"Error al eliminar: {e}")
    connection.rollback()

Eliminación con múltiples condiciones

Podemos usar condiciones más complejas para eliminar registros:

try:
    # Eliminar registros con múltiples condiciones
    query = """
    DELETE FROM empleados 
    WHERE departamento = %s AND fecha_contratacion < %s
    """
    
    departamento = "Temporal"
    fecha_limite = "2023-01-01"
    
    cursor.execute(query, (departamento, fecha_limite))
    
    connection.commit()
    
    print(f"Se eliminaron {cursor.rowcount} empleados temporales contratados antes de {fecha_limite}")
    
except Error as e:
    print(f"Error al eliminar con condiciones: {e}")
    connection.rollback()

Manejo de errores específicos en operaciones CRUD

Es importante manejar adecuadamente los errores que pueden ocurrir durante las operaciones CRUD:

try:
    # Intentar insertar un registro que podría violar una restricción
    query = "INSERT INTO empleados (id, nombre, email) VALUES (%s, %s, %s)"
    valores = (1, "Juan Duplicado", "juan@ejemplo.com")  # ID posiblemente duplicado
    
    cursor.execute(query, valores)
    connection.commit()
    
except mysql.connector.errors.IntegrityError as e:
    if e.errno == 1062:  # Código de error para duplicado
        print("Error: No se puede insertar un registro duplicado")
    elif e.errno == 1452:  # Código para violación de clave foránea
        print("Error: Violación de restricción de clave foránea")
    else:
        print(f"Error de integridad: {e}")
    connection.rollback()
    
except mysql.connector.errors.DataError as e:
    print(f"Error en los datos (tipo o tamaño incorrecto): {e}")
    connection.rollback()
    
except Error as e:
    print(f"Error general: {e}")
    connection.rollback()

Optimización de operaciones CRUD

Para mejorar el rendimiento de las operaciones CRUD, podemos aplicar algunas técnicas:

Uso de transacciones para operaciones por lotes

try:
    # Desactivar autocommit para controlar manualmente la transacción
    connection.autocommit = False
    
    # Realizar múltiples operaciones en una sola transacción
    cursor.execute("DELETE FROM registros_temporales WHERE fecha < %s", ("2023-01-01",))
    cursor.execute("UPDATE estadisticas SET total = total - %s", (cursor.rowcount,))
    cursor.execute("INSERT INTO log_operaciones (tipo, registros_afectados) VALUES (%s, %s)", 
                  ("limpieza", cursor.rowcount))
    
    # Confirmar todas las operaciones de una vez
    connection.commit()
    print("Transacción completada con éxito")
    
except Error as e:
    print(f"Error en la transacción: {e}")
    connection.rollback()
    print("Se ha revertido la transacción")
    
finally:
    # Restaurar el comportamiento predeterminado
    connection.autocommit = True

Las transacciones son especialmente útiles cuando necesitamos garantizar que un conjunto de operaciones se complete en su totalidad o no se realice ninguna.

Uso de consultas optimizadas

# En lugar de múltiples consultas individuales
try:
    # Consulta optimizada con JOIN
    query = """
    SELECT e.nombre, e.apellido, d.nombre_departamento, d.ubicacion
    FROM empleados e
    INNER JOIN departamentos d ON e.departamento_id = d.id
    WHERE d.ubicacion = %s
    """
    
    cursor.execute(query, ("Madrid",))
    resultados = cursor.fetchall()
    
    # Procesar resultados...
    
except Error as e:
    print(f"Error en consulta optimizada: {e}")

Ejemplo completo de operaciones CRUD

A continuación, un ejemplo que integra las cuatro operaciones CRUD en un flujo de trabajo coherente:

import mysql.connector
from mysql.connector import Error

def ejecutar_operaciones_crud():
    try:
        # Establecer conexión
        connection = mysql.connector.connect(
            host="localhost",
            user="usuario",
            password="contraseña",
            database="empresa"
        )
        
        cursor = connection.cursor()
        
        # 1. CREATE - Insertar un nuevo producto
        print("\n--- Operación CREATE ---")
        insert_query = """
        INSERT INTO productos (nombre, descripcion, precio, stock)
        VALUES (%s, %s, %s, %s)
        """
        producto_nuevo = ("Teclado Ergonómico", "Teclado con diseño ergonómico para programadores", 89.99, 15)
        cursor.execute(insert_query, producto_nuevo)
        connection.commit()
        
        id_producto = cursor.lastrowid
        print(f"Producto insertado con ID: {id_producto}")
        
        # 2. READ - Leer el producto insertado
        print("\n--- Operación READ ---")
        cursor.execute("SELECT * FROM productos WHERE id = %s", (id_producto,))
        producto = cursor.fetchone()
        print(f"Producto leído: {producto}")
        
        # 3. UPDATE - Actualizar el stock del producto
        print("\n--- Operación UPDATE ---")
        cursor.execute("UPDATE productos SET stock = stock + 5 WHERE id = %s", (id_producto,))
        connection.commit()
        print(f"Stock actualizado. Filas afectadas: {cursor.rowcount}")
        
        # Verificar el cambio
        cursor.execute("SELECT nombre, stock FROM productos WHERE id = %s", (id_producto,))
        producto_actualizado = cursor.fetchone()
        print(f"Producto '{producto_actualizado[0]}' ahora tiene {producto_actualizado[1]} unidades en stock")
        
        # 4. DELETE - Eliminar el producto (comentado para no perder el registro)
        print("\n--- Operación DELETE ---")
        # cursor.execute("DELETE FROM productos WHERE id = %s", (id_producto,))
        # connection.commit()
        # print(f"Producto eliminado. Filas afectadas: {cursor.rowcount}")
        
        print("\nSimulación de eliminación (no ejecutada para preservar el ejemplo)")
        
    except Error as e:
        print(f"Error en las operaciones CRUD: {e}")
        if 'connection' in locals() and connection.is_connected():
            connection.rollback()
    
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()
            print("\nConexión cerrada")

# Ejecutar el ejemplo
if __name__ == "__main__":
    ejecutar_operaciones_crud()

Este ejemplo muestra cómo las operaciones CRUD forman un ciclo completo de gestión de datos, desde la creación hasta la eliminación de registros, pasando por la lectura y actualización.

Consultas parametrizadas

Las consultas parametrizadas son una técnica fundamental para interactuar con bases de datos de manera segura y eficiente. A diferencia de las consultas directas, donde los valores se concatenan a la cadena SQL, las consultas parametrizadas utilizan marcadores de posición que se sustituyen por valores reales durante la ejecución.

Importancia de las consultas parametrizadas

Cuando trabajamos con MySQL desde Python, el uso de consultas parametrizadas ofrece dos ventajas principales:

Seguridad: Protegen contra ataques de inyección SQL al separar el código SQL de los datos.
Rendimiento: Permiten la reutilización de consultas preparadas, mejorando la eficiencia.

Sintaxis básica

MySQL Connector/Python soporta dos estilos de parametrización:

import mysql.connector

connection = mysql.connector.connect(
    host="localhost",
    user="usuario",
    password="contraseña",
    database="mi_base"
)

cursor = connection.cursor()

# Estilo con %s (recomendado)
cursor.execute("SELECT * FROM usuarios WHERE edad > %s AND ciudad = %s", (25, "Madrid"))

# Estilo con formato de diccionario
cursor.execute("SELECT * FROM usuarios WHERE edad > %(edad)s AND ciudad = %(ciudad)s", 
               {'edad': 25, 'ciudad': 'Madrid'})

El estilo con %s es más conciso, mientras que el estilo con diccionario mejora la legibilidad cuando hay muchos parámetros.

Tipos de datos y conversión automática

MySQL Connector maneja automáticamente la conversión entre tipos de datos Python y MySQL:

# Diferentes tipos de datos
nombre = "Ana García"       # String -> VARCHAR/TEXT
edad = 28                   # Integer -> INT
salario = 45000.50          # Float -> DECIMAL
activo = True               # Boolean -> TINYINT(1)
fecha_inicio = datetime.date(2023, 5, 15)  # Date -> DATE

query = """
INSERT INTO empleados (nombre, edad, salario, activo, fecha_inicio) 
VALUES (%s, %s, %s, %s, %s)
"""

cursor.execute(query, (nombre, edad, salario, activo, fecha_inicio))
connection.commit()

El conector se encarga de escapar caracteres especiales y formatear correctamente cada tipo de dato según los requisitos de MySQL.

Consultas SELECT parametrizadas

Las consultas de selección parametrizadas son especialmente útiles para filtrar resultados:

def buscar_productos(categoria, precio_min, disponible=True):
    try:
        query = """
        SELECT id, nombre, precio, stock
        FROM productos
        WHERE categoria = %s 
        AND precio >= %s
        AND disponible = %s
        ORDER BY precio
        """
        
        cursor.execute(query, (categoria, precio_min, disponible))
        
        productos = cursor.fetchall()
        
        print(f"Productos de {categoria} con precio >= {precio_min}:")
        for producto in productos:
            print(f"ID: {producto[0]}, Nombre: {producto[1]}, Precio: {producto[2]}€, Stock: {producto[3]}")
            
        return productos
        
    except mysql.connector.Error as err:
        print(f"Error en la consulta: {err}")
        return []

# Uso de la función
buscar_productos("Electrónica", 300)

Esta función permite reutilizar la misma lógica con diferentes parámetros, manteniendo el código limpio y seguro.

Consultas parametrizadas con LIKE

Para búsquedas por patrones, debemos tener cuidado con la sintaxis de LIKE:

def buscar_por_nombre(patron):
    try:
        # El % debe ser parte del parámetro, no de la consulta SQL
        patron_busqueda = f"%{patron}%"
        
        query = "SELECT id, nombre, email FROM clientes WHERE nombre LIKE %s"
        cursor.execute(query, (patron_busqueda,))
        
        resultados = cursor.fetchall()
        print(f"Se encontraron {len(resultados)} clientes que coinciden con '{patron}'")
        
        for cliente in resultados:
            print(f"{cliente[0]} - {cliente[1]} ({cliente[2]})")
            
    except mysql.connector.Error as err:
        print(f"Error en búsqueda: {err}")

# Ejemplos de uso
buscar_por_nombre("Mart")  # Encontrará "Martínez", "Martín", etc.

Es importante notar que los caracteres comodín (% y _) deben ser parte del valor del parámetro, no de la consulta SQL.

Consultas parametrizadas con IN

Para consultas que utilizan la cláusula IN, necesitamos un enfoque especial:

def obtener_productos_por_ids(lista_ids):
    try:
        # Crear marcadores de posición dinámicos según la cantidad de IDs
        placeholders = ', '.join(['%s'] * len(lista_ids))
        
        query = f"SELECT id, nombre, precio FROM productos WHERE id IN ({placeholders})"
        
        # Convertir lista a tupla para cursor.execute
        cursor.execute(query, tuple(lista_ids))
        
        productos = cursor.fetchall()
        
        for producto in productos:
            print(f"ID: {producto[0]}, Nombre: {producto[1]}, Precio: {producto[2]}€")
            
    except mysql.connector.Error as err:
        print(f"Error al obtener productos: {err}")

# Ejemplo de uso
obtener_productos_por_ids([1, 5, 9, 12])

Aquí generamos dinámicamente los marcadores de posición según el número de elementos en la lista, manteniendo la seguridad de la consulta parametrizada.

Consultas parametrizadas para operaciones de actualización

Las consultas parametrizadas son igualmente importantes para operaciones de escritura:

def actualizar_stock_producto(producto_id, cantidad, operacion="sumar"):
    try:
        if operacion.lower() == "sumar":
            query = "UPDATE productos SET stock = stock + %s WHERE id = %s"
        elif operacion.lower() == "restar":
            query = "UPDATE productos SET stock = stock - %s WHERE id = %s"
        else:
            query = "UPDATE productos SET stock = %s WHERE id = %s"
        
        cursor.execute(query, (cantidad, producto_id))
        connection.commit()
        
        if cursor.rowcount > 0:
            print(f"Stock actualizado para producto ID {producto_id}")
            
            # Consultar el stock actual
            cursor.execute("SELECT nombre, stock FROM productos WHERE id = %s", (producto_id,))
            producto = cursor.fetchone()
            print(f"Producto '{producto[0]}' - Stock actual: {producto[1]} unidades")
            return True
        else:
            print(f"No se encontró el producto con ID {producto_id}")
            return False
            
    except mysql.connector.Error as err:
        print(f"Error al actualizar stock: {err}")
        connection.rollback()
        return False

# Ejemplos de uso
actualizar_stock_producto(5, 10, "sumar")    # Aumentar stock
actualizar_stock_producto(8, 3, "restar")    # Disminuir stock
actualizar_stock_producto(12, 25, "fijar")   # Establecer valor exacto

Esta función demuestra cómo las consultas parametrizadas pueden hacer que el código sea más flexible y reutilizable.

Consultas parametrizadas con procedimientos almacenados

MySQL Connector también permite ejecutar procedimientos almacenados con parámetros:

def ejecutar_procedimiento_venta(cliente_id, producto_id, cantidad):
    try:
        # Llamar a un procedimiento almacenado con parámetros
        cursor.callproc('realizar_venta', (cliente_id, producto_id, cantidad))
        
        # Obtener resultados si el procedimiento devuelve algo
        for result in cursor.stored_results():
            for row in result.fetchall():
                print(row)
        
        # Confirmar la transacción
        connection.commit()
        print("Venta procesada correctamente")
        return True
        
    except mysql.connector.Error as err:
        print(f"Error al ejecutar procedimiento: {err}")
        connection.rollback()
        return False

# Ejemplo de uso
ejecutar_procedimiento_venta(42, 7, 2)

Los procedimientos almacenados pueden mejorar significativamente el rendimiento y la seguridad al mover la lógica compleja al servidor de base de datos.

Manejo de errores específicos en consultas parametrizadas

Es importante manejar adecuadamente los errores que pueden surgir:

def insertar_usuario(nombre, email, edad):
    try:
        query = "INSERT INTO usuarios (nombre, email, edad) VALUES (%s, %s, %s)"
        cursor.execute(query, (nombre, email, edad))
        connection.commit()
        return cursor.lastrowid
        
    except mysql.connector.errors.IntegrityError as err:
        if err.errno == 1062:  # Código de error para duplicado
            print(f"Error: El email '{email}' ya está registrado")
        else:
            print(f"Error de integridad: {err}")
        connection.rollback()
        return None
        
    except mysql.connector.errors.DataError as err:
        print(f"Error en los datos proporcionados: {err}")
        connection.rollback()
        return None
        
    except mysql.connector.Error as err:
        print(f"Error en la base de datos: {err}")
        connection.rollback()
        return None

# Probar con datos válidos e inválidos
nuevo_id = insertar_usuario("Elena López", "elena@ejemplo.com", 29)
if nuevo_id:
    print(f"Usuario creado con ID: {nuevo_id}")

# Intentar con email duplicado
insertar_usuario("Elena Duplicada", "elena@ejemplo.com", 35)

Este enfoque permite proporcionar mensajes de error específicos según el tipo de problema encontrado.

Consultas parametrizadas con valores opcionales

A veces necesitamos construir consultas donde algunos filtros son opcionales:

def buscar_empleados(departamento=None, ciudad=None, salario_min=None):
    try:
        # Comenzar con la parte fija de la consulta
        query = "SELECT id, nombre, departamento, ciudad, salario FROM empleados WHERE 1=1"
        params = []
        
        # Añadir condiciones según los parámetros proporcionados
        if departamento:
            query += " AND departamento = %s"
            params.append(departamento)
            
        if ciudad:
            query += " AND ciudad = %s"
            params.append(ciudad)
            
        if salario_min:
            query += " AND salario >= %s"
            params.append(salario_min)
        
        # Ordenar resultados
        query += " ORDER BY nombre"
        
        # Ejecutar la consulta con los parámetros acumulados
        cursor.execute(query, tuple(params))
        
        empleados = cursor.fetchall()
        print(f"Se encontraron {len(empleados)} empleados")
        
        for emp in empleados:
            print(f"{emp[1]} - {emp[2]} ({emp[3]}) - {emp[4]}€")
            
        return empleados
        
    except mysql.connector.Error as err:
        print(f"Error en la búsqueda: {err}")
        return []

# Ejemplos de uso con diferentes combinaciones de parámetros
buscar_empleados(departamento="Ventas")
buscar_empleados(ciudad="Barcelona")
buscar_empleados(salario_min=40000)
buscar_empleados(departamento="Desarrollo", salario_min=45000)

Esta técnica permite crear consultas dinámicas manteniendo la seguridad de las consultas parametrizadas.

Mejores prácticas para consultas parametrizadas

Para aprovechar al máximo las consultas parametrizadas, sigue estas recomendaciones:

Nunca concatenes valores directamente en las consultas SQL.
Usa marcadores de posición para todos los valores variables.
Mantén separada la lógica de construcción de consultas de los datos.
Utiliza el estilo de parámetros adecuado según la complejidad de tu consulta.
Implementa manejo de errores específico para diferentes tipos de problemas.
Considera usar procedimientos almacenados para operaciones complejas.

# ❌ Forma incorrecta (vulnerable a inyección SQL)
nombre_usuario = "usuario' OR '1'='1"
cursor.execute(f"SELECT * FROM usuarios WHERE nombre = '{nombre_usuario}'")

# ✅ Forma correcta (consulta parametrizada)
cursor.execute("SELECT * FROM usuarios WHERE nombre = %s", (nombre_usuario,))

Siguiendo estas prácticas, tus aplicaciones serán más seguras, mantenibles y eficientes al interactuar con bases de datos MySQL.

Manejo de transacciones

Las transacciones son un mecanismo fundamental en los sistemas de bases de datos que garantiza la integridad de los datos al agrupar operaciones relacionadas en una unidad lógica de trabajo. En MySQL, las transacciones permiten que múltiples operaciones se ejecuten como una sola unidad atómica, asegurando que o bien todas las operaciones se completan con éxito, o ninguna de ellas se aplica.

Principios ACID de las transacciones

Las transacciones en MySQL siguen los principios ACID:

Atomicidad: Todas las operaciones dentro de una transacción se ejecutan completamente o ninguna se ejecuta.
Consistencia: La base de datos pasa de un estado válido a otro estado válido.
Aislamiento: Las transacciones se ejecutan de forma aislada, sin interferir entre sí.
Durabilidad: Una vez confirmada una transacción, los cambios persisten incluso ante fallos del sistema.

Control de transacciones con MySQL Connector

MySQL Connector/Python proporciona métodos para controlar transacciones de forma explícita:

import mysql.connector
from mysql.connector import Error

try:
    connection = mysql.connector.connect(
        host="localhost",
        user="usuario",
        password="contraseña",
        database="mi_base"
    )
    
    # Por defecto, autocommit está desactivado
    print(f"Autocommit: {connection.autocommit}")
    
    cursor = connection.cursor()
    
    # Iniciar una transacción explícitamente
    cursor.execute("START TRANSACTION")
    
    # Realizar operaciones dentro de la transacción
    cursor.execute("UPDATE cuentas SET saldo = saldo - 1000 WHERE id = 1")
    cursor.execute("UPDATE cuentas SET saldo = saldo + 1000 WHERE id = 2")
    
    # Confirmar la transacción
    connection.commit()
    print("Transacción completada con éxito")
    
except Error as e:
    # Revertir la transacción en caso de error
    connection.rollback()
    print(f"Error: {e}")
    print("Transacción revertida")
    
finally:
    if connection.is_connected():
        cursor.close()
        connection.close()

En este ejemplo, ambas operaciones de actualización forman parte de una única transacción. Si cualquiera de ellas falla, la llamada a rollback() deshace todos los cambios, manteniendo la consistencia de los datos.

Control de autocommit

MySQL Connector permite controlar el comportamiento de autocommit:

# Desactivar autocommit (comportamiento predeterminado)
connection.autocommit = False

# Activar autocommit (cada operación se confirma inmediatamente)
connection.autocommit = True

Cuando autocommit está activado, cada operación SQL se confirma automáticamente. Esto es útil para consultas individuales, pero para transacciones que involucran múltiples operaciones, es mejor mantenerlo desactivado.

Uso de context managers para transacciones

Python 3 permite utilizar el patrón de contexto para gestionar transacciones de forma más elegante:

def transferir_fondos(origen_id, destino_id, cantidad):
    try:
        with mysql.connector.connect(**config) as connection:
            # Desactivar autocommit explícitamente
            connection.autocommit = False
            
            with connection.cursor() as cursor:
                # Verificar saldo suficiente
                cursor.execute("SELECT saldo FROM cuentas WHERE id = %s", (origen_id,))
                saldo_actual = cursor.fetchone()[0]
                
                if saldo_actual < cantidad:
                    raise ValueError("Saldo insuficiente para realizar la transferencia")
                
                # Realizar la transferencia
                cursor.execute(
                    "UPDATE cuentas SET saldo = saldo - %s WHERE id = %s",
                    (cantidad, origen_id)
                )
                
                cursor.execute(
                    "UPDATE cuentas SET saldo = saldo + %s WHERE id = %s",
                    (cantidad, destino_id)
                )
                
                # Registrar la transacción
                cursor.execute(
                    "INSERT INTO movimientos (origen_id, destino_id, cantidad, fecha) VALUES (%s, %s, %s, NOW())",
                    (origen_id, destino_id, cantidad)
                )
                
                # Si llegamos aquí sin errores, confirmamos la transacción
                connection.commit()
                print(f"Transferencia de {cantidad}€ completada con éxito")
                return True
                
    except Error as e:
        if 'connection' in locals() and connection.is_connected():
            connection.rollback()
        print(f"Error en la transferencia: {e}")
        return False
    except ValueError as e:
        print(f"Error de validación: {e}")
        return False

Este patrón garantiza que los recursos se liberen correctamente y que la transacción se revierta automáticamente si ocurre una excepción.

Puntos de guardado (savepoints)

Para transacciones complejas, MySQL permite establecer puntos de guardado que permiten revertir parcialmente una transacción:

try:
    connection.autocommit = False
    cursor = connection.cursor()
    
    # Iniciar la transacción
    cursor.execute("START TRANSACTION")
    
    # Primera operación
    cursor.execute("INSERT INTO pedidos (cliente_id, fecha) VALUES (%s, NOW())", (cliente_id,))
    pedido_id = cursor.lastrowid
    
    # Establecer un punto de guardado después de crear el pedido
    cursor.execute("SAVEPOINT pedido_creado")
    
    try:
        # Añadir productos al pedido
        for producto_id, cantidad in productos:
            # Verificar stock disponible
            cursor.execute("SELECT stock FROM productos WHERE id = %s", (producto_id,))
            stock_disponible = cursor.fetchone()[0]
            
            if stock_disponible < cantidad:
                # Revertir hasta el punto de guardado
                cursor.execute("ROLLBACK TO SAVEPOINT pedido_creado")
                raise ValueError(f"Stock insuficiente para el producto {producto_id}")
            
            # Añadir línea de pedido
            cursor.execute(
                "INSERT INTO pedido_detalles (pedido_id, producto_id, cantidad) VALUES (%s, %s, %s)",
                (pedido_id, producto_id, cantidad)
            )
            
            # Actualizar stock
            cursor.execute(
                "UPDATE productos SET stock = stock - %s WHERE id = %s",
                (cantidad, producto_id)
            )
        
        # Confirmar el pedido completo
        connection.commit()
        print(f"Pedido {pedido_id} procesado correctamente")
        
    except ValueError as e:
        # Ya hemos revertido hasta el punto de guardado, ahora completamos el pedido sin los productos problemáticos
        connection.commit()
        print(f"Pedido {pedido_id} procesado parcialmente: {e}")
        
except Error as e:
    connection.rollback()
    print(f"Error en el procesamiento del pedido: {e}")

Los puntos de guardado son especialmente útiles cuando necesitas control granular sobre qué partes de una transacción deben revertirse en caso de error.

Niveles de aislamiento de transacciones

MySQL permite configurar diferentes niveles de aislamiento para controlar cómo interactúan las transacciones concurrentes:

try:
    connection = mysql.connector.connect(**config)
    cursor = connection.cursor()
    
    # Consultar el nivel de aislamiento actual
    cursor.execute("SELECT @@transaction_isolation")
    nivel_actual = cursor.fetchone()[0]
    print(f"Nivel de aislamiento actual: {nivel_actual}")
    
    # Cambiar el nivel de aislamiento para esta sesión
    cursor.execute("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")
    
    # Iniciar una transacción con el nuevo nivel
    cursor.execute("START TRANSACTION")
    
    # Realizar operaciones...
    
    connection.commit()
    
except Error as e:
    connection.rollback()
    print(f"Error: {e}")

Los niveles de aislamiento disponibles en MySQL son:

READ UNCOMMITTED: Permite leer cambios no confirmados (puede causar lecturas sucias).
READ COMMITTED: Solo permite leer cambios confirmados (evita lecturas sucias).
REPEATABLE READ: Garantiza que las lecturas repetidas den el mismo resultado (predeterminado en MySQL).
SERIALIZABLE: El nivel más estricto, que evita todos los problemas de concurrencia.

La elección del nivel adecuado depende del equilibrio entre consistencia y rendimiento que requiera tu aplicación.

Manejo de bloqueos y tiempos de espera

En entornos con alta concurrencia, los bloqueos pueden afectar el rendimiento. MySQL Connector permite configurar tiempos de espera para las transacciones:

try:
    connection = mysql.connector.connect(**config)
    cursor = connection.cursor()
    
    # Establecer un tiempo de espera para bloqueos (en segundos)
    cursor.execute("SET innodb_lock_wait_timeout = 10")
    
    # Iniciar la transacción
    cursor.execute("START TRANSACTION")
    
    # Intentar actualizar una fila que podría estar bloqueada
    cursor.execute("UPDATE productos SET stock = stock - 1 WHERE id = %s FOR UPDATE", (producto_id,))
    
    # Resto de operaciones...
    
    connection.commit()
    
except mysql.connector.errors.DatabaseError as e:
    if "lock wait timeout exceeded" in str(e).lower():
        print("La operación no pudo completarse porque los recursos están bloqueados")
    else:
        print(f"Error de base de datos: {e}")
    connection.rollback()

La cláusula FOR UPDATE adquiere un bloqueo exclusivo sobre las filas seleccionadas, y el tiempo de espera configurado determina cuánto tiempo esperar antes de abandonar si otra transacción ya tiene el bloqueo.

Transacciones distribuidas y XA

Para aplicaciones que interactúan con múltiples bases de datos, MySQL soporta transacciones XA (eXtended Architecture):

try:
    # Conexiones a diferentes bases de datos
    conn1 = mysql.connector.connect(**config_db1)
    conn2 = mysql.connector.connect(**config_db2)
    
    cursor1 = conn1.cursor()
    cursor2 = conn2.cursor()
    
    # Generar un ID de transacción único
    xid = f"xatest-{uuid.uuid4().hex}"
    
    # Iniciar transacción XA en ambas conexiones
    cursor1.execute(f"XA START '{xid}'")
    cursor2.execute(f"XA START '{xid}'")
    
    # Realizar operaciones en la primera base de datos
    cursor1.execute("UPDATE cuentas_db1 SET saldo = saldo - %s WHERE id = %s", (cantidad, cuenta1))
    
    # Realizar operaciones en la segunda base de datos
    cursor2.execute("UPDATE cuentas_db2 SET saldo = saldo + %s WHERE id = %s", (cantidad, cuenta2))
    
    # Preparar la transacción en ambas conexiones
    cursor1.execute(f"XA END '{xid}'")
    cursor2.execute(f"XA END '{xid}'")
    
    cursor1.execute(f"XA PREPARE '{xid}'")
    cursor2.execute(f"XA PREPARE '{xid}'")
    
    # Fase de commit
    try:
        cursor1.execute(f"XA COMMIT '{xid}'")
        cursor2.execute(f"XA COMMIT '{xid}'")
        print("Transacción XA completada con éxito")
    except:
        # Si falla el commit en cualquier base de datos, intentamos revertir en ambas
        try:
            cursor1.execute(f"XA ROLLBACK '{xid}'")
            cursor2.execute(f"XA ROLLBACK '{xid}'")
        except:
            print("Error crítico: transacción en estado inconsistente")
        raise
    
except Error as e:
    print(f"Error en transacción XA: {e}")
    # Intentar rollback si es posible

Las transacciones XA son más complejas pero garantizan la consistencia entre múltiples recursos transaccionales.

Patrones de diseño para transacciones

Patrón de repositorio con transacciones

Un enfoque estructurado para manejar transacciones es implementar un patrón de repositorio:

class ClienteRepository:
    def __init__(self, connection):
        self.connection = connection
        self.cursor = connection.cursor(dictionary=True)
    
    def crear_con_direcciones(self, cliente_data, direcciones):
        try:
            # Iniciar transacción
            self.connection.start_transaction()
            
            # Insertar cliente
            query = """
            INSERT INTO clientes (nombre, email, telefono)
            VALUES (%s, %s, %s)
            """
            self.cursor.execute(query, (
                cliente_data['nombre'],
                cliente_data['email'],
                cliente_data['telefono']
            ))
            
            cliente_id = self.cursor.lastrowid
            
            # Insertar direcciones
            for direccion in direcciones:
                query = """
                INSERT INTO direcciones (cliente_id, tipo, calle, ciudad, codigo_postal)
                VALUES (%s, %s, %s, %s, %s)
                """
                self.cursor.execute(query, (
                    cliente_id,
                    direccion['tipo'],
                    direccion['calle'],
                    direccion['ciudad'],
                    direccion['codigo_postal']
                ))
            
            # Confirmar transacción
            self.connection.commit()
            return cliente_id
            
        except Error as e:
            self.connection.rollback()
            raise Exception(f"Error al crear cliente: {e}")

Este patrón encapsula la lógica de negocio y el manejo de transacciones, facilitando la reutilización y el mantenimiento del código.

Patrón Unit of Work

Para operaciones más complejas que involucran múltiples repositorios, el patrón Unit of Work proporciona un control centralizado de transacciones:

class UnitOfWork:
    def __init__(self, connection_params):
        self.connection_params = connection_params
        self.connection = None
    
    def __enter__(self):
        self.connection = mysql.connector.connect(**self.connection_params)
        self.connection.autocommit = False
        
        # Crear repositorios con la misma conexión
        self.clientes = ClienteRepository(self.connection)
        self.pedidos = PedidoRepository(self.connection)
        self.productos = ProductoRepository(self.connection)
        
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is not None:
            # Si hay una excepción, revertir
            self.connection.rollback()
        
        if self.connection.is_connected():
            self.connection.close()
    
    def commit(self):
        self.connection.commit()

Uso del patrón Unit of Work:

try:
    with UnitOfWork(config) as uow:
        # Todas estas operaciones comparten la misma transacción
        cliente_id = uow.clientes.crear({"nombre": "Juan Pérez", "email": "juan@ejemplo.com"})
        producto_ids = uow.productos.buscar_por_categoria("Electrónica")
        pedido_id = uow.pedidos.crear(cliente_id, producto_ids)
        
        # Confirmar todas las operaciones de una vez
        uow.commit()
        print("Operación completada con éxito")
        
except Exception as e:
    print(f"Error: {e}")
    # No es necesario hacer rollback explícito, lo maneja el context manager

Este patrón garantiza que todas las operaciones relacionadas se ejecuten en la misma transacción, manteniendo la integridad referencial entre entidades.

Monitoreo y depuración de transacciones

Para aplicaciones críticas, es útil implementar un sistema de monitoreo de transacciones:

def ejecutar_transaccion(operacion, parametros):
    start_time = time.time()
    transaccion_id = str(uuid.uuid4())
    
    log_entry = {
        "id": transaccion_id,
        "operacion": operacion,
        "inicio": datetime.now().isoformat(),
        "estado": "iniciada"
    }
    
    try:
        connection = mysql.connector.connect(**config)
        connection.autocommit = False
        cursor = connection.cursor()
        
        # Registrar inicio de transacción
        print(f"[{transaccion_id}] Iniciando transacción: {operacion}")
        
        # Ejecutar operaciones
        resultado = operacion(connection, cursor, parametros)
        
        # Confirmar transacción
        connection.commit()
        
        # Actualizar registro
        log_entry["estado"] = "completada"
        log_entry["duracion"] = time.time() - start_time
        print(f"[{transaccion_id}] Transacción completada en {log_entry['duracion']:.2f} segundos")
        
        return resultado
        
    except Exception as e:
        # Registrar error
        log_entry["estado"] = "fallida"
        log_entry["error"] = str(e)
        log_entry["duracion"] = time.time() - start_time
        
        print(f"[{transaccion_id}] Error en transacción: {e}")
        
        if 'connection' in locals() and connection.is_connected():
            connection.rollback()
            
        raise
        
    finally:
        if 'connection' in locals() and connection.is_connected():
            cursor.close()
            connection.close()
        
        # Guardar registro de la transacción (en archivo, base de datos, etc.)
        with open("transacciones.log", "a") as log_file:
            log_file.write(json.dumps(log_entry) + "\n")

Este sistema de monitoreo proporciona visibilidad sobre el rendimiento y los errores de las transacciones, facilitando la depuración y optimización.

Consideraciones de rendimiento

Al trabajar con transacciones en MySQL, es importante considerar estos aspectos de rendimiento:

Duración de las transacciones: Mantén las transacciones lo más cortas posible para reducir el tiempo de bloqueo.
Tamaño de las transacciones: Evita transacciones que afecten a un gran número de filas.
Orden de las operaciones: Realiza primero las operaciones de lectura y luego las de escritura para minimizar bloqueos.
Índices adecuados: Asegúrate de que las tablas involucradas tengan índices apropiados.

# Ejemplo de transacción optimizada
def procesar_lote(registros, tamano_lote=100):
    connection = mysql.connector.connect(**config)
    connection.autocommit = False
    cursor = connection.cursor()
    
    try:
        total_procesados = 0
        
        # Procesar en lotes para evitar transacciones demasiado grandes
        for i in range(0, len(registros), tamano_lote):
            lote_actual = registros[i:i+tamano_lote]
            
            # Iniciar transacción para este lote
            cursor.execute("START TRANSACTION")
            
            for registro in lote_actual:
                # Procesar cada registro
                cursor.execute(
                    "INSERT INTO procesados (id, datos, fecha) VALUES (%s, %s, NOW())",
                    (registro['id'], registro['datos'])
                )
            
            # Confirmar este lote
            connection.commit()
            total_procesados += len(lote_actual)
            print(f"Lote procesado: {total_procesados}/{len(registros)}")
        
        return total_procesados
        
    except Error as e:
        connection.rollback()
        print(f"Error: {e}")
        return total_procesados
        
    finally:
        cursor.close()
        connection.close()

Este enfoque de procesamiento por lotes equilibra la atomicidad con el rendimiento, evitando transacciones excesivamente largas.

Las transacciones son una herramienta poderosa para mantener la integridad de los datos en aplicaciones que utilizan MySQL. Cuando se implementan correctamente, garantizan que las operaciones complejas se ejecuten de manera segura y consistente, incluso en entornos con alta concurrencia o ante fallos del sistema.

Aprende Python online

Otros ejercicios de programación de Python

Evalúa tus conocimientos de esta lección Acceso a MySQL con MySQL Connector con nuestros retos de programación de tipo Test, Puzzle, Código y Proyecto con VSCode, guiados por IA.

Módulo math