DELETE en Python

Eliminar registros de MySQL desde Python es una operación que debe realizarse con precaución, ya que los datos eliminados de forma permanente no se pueden recuperar fácilmente. En este artículo aprenderás a ejecutar sentencias DELETE de forma segura, implementar transacciones con rollback para proteger la integridad de los datos y aplicar patrones como el borrado lógico que permiten una gestión más flexible de la información.

Requisitos previos

Necesitas una conexión configurada y las tablas de prueba con datos. Usa el siguiente SQL si necesitas crearlas:

USE tienda;
 
CREATE TABLE IF NOT EXISTS productos (
    id INT AUTO_INCREMENT PRIMARY KEY,
    nombre VARCHAR(100) NOT NULL,
    categoria VARCHAR(50) NOT NULL,
    precio DECIMAL(10, 2) NOT NULL,
    stock INT DEFAULT 0,
    activo TINYINT(1) DEFAULT 1,
    eliminado_en DATETIME DEFAULT NULL,
    fecha_creacion DATETIME DEFAULT CURRENT_TIMESTAMP
);
 
INSERT INTO productos (nombre, categoria, precio, stock) VALUES
('Laptop HP Pavilion', 'Computadoras', 12999.99, 25),
('Mouse Logitech MX Master', 'Periféricos', 1599.00, 150),
('Teclado Mecánico Corsair K70', 'Periféricos', 2299.50, 80),
('Monitor Samsung 27"', 'Monitores', 6499.00, 40),
('Auriculares Sony WH-1000XM5', 'Audio', 5999.99, 60);

Código completo

Este ejemplo elimina un producto por su ID:

import mysql.connector
 
def eliminar_producto():
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda'
    )
 
    producto_id = 5
    cursor = conexion.cursor()
    cursor.execute('DELETE FROM productos WHERE id = %s', (producto_id,))
    conexion.commit()
 
    if cursor.rowcount > 0:
        print(f'Producto con ID {producto_id} eliminado correctamente')
        print(f'Filas eliminadas: {cursor.rowcount}')
    else:
        print(f'No se encontró el producto con ID {producto_id}')
 
    cursor.close()
    conexion.close()
 
eliminar_producto()

Salida esperada:

Producto con ID 5 eliminado correctamente
Filas eliminadas: 1

Explicación paso a paso

La sentencia DELETE se ejecuta de la misma forma que cualquier otra sentencia SQL en Python. El parámetro se pasa como una tupla con un solo elemento (nota la coma después del valor), y cursor.rowcount indica cuántas filas se eliminaron. Es fundamental siempre incluir una cláusula WHERE para evitar eliminar todos los registros de la tabla.

El llamado a conexion.commit() es necesario para confirmar la eliminación en la base de datos. Si no haces commit, los cambios se perderán cuando se cierre la conexión. Esto es útil porque te permite hacer rollback si algo sale mal antes de confirmar.

Eliminar con verificación previa

Una práctica segura es verificar qué se va a eliminar antes de hacerlo:

def eliminar_con_verificacion(producto_id):
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda'
    )
 
    cursor = conexion.cursor(dictionary=True)
 
    # Verificar que el producto existe
    cursor.execute(
        'SELECT id, nombre, precio FROM productos WHERE id = %s',
        (producto_id,)
    )
    producto = cursor.fetchone()
 
    if not producto:
        print('Producto no encontrado')
        cursor.close()
        conexion.close()
        return False
 
    print(f"Eliminando: {producto['nombre']} (${producto['precio']})")
 
    cursor.execute('DELETE FROM productos WHERE id = %s', (producto_id,))
    conexion.commit()
 
    print('Producto eliminado correctamente')
    cursor.close()
    conexion.close()
    return True

Transacción con rollback

Cuando eliminas datos que están relacionados con otras tablas, es importante usar transacciones para garantizar la consistencia:

def eliminar_categoria_completa(categoria):
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda',
        autocommit=False
    )
 
    try:
        cursor = conexion.cursor(dictionary=True)
 
        # Contar productos afectados
        cursor.execute(
            'SELECT COUNT(*) AS total FROM productos WHERE categoria = %s',
            (categoria,)
        )
        total = cursor.fetchone()['total']
 
        if total == 0:
            print(f'No hay productos en la categoría "{categoria}"')
            return 0
 
        # Eliminar productos de la categoría
        cursor.execute(
            'DELETE FROM productos WHERE categoria = %s',
            (categoria,)
        )
        eliminados = cursor.rowcount
 
        conexion.commit()
        print(f'{eliminados} productos eliminados de la categoría "{categoria}"')
        return eliminados
 
    except Exception as error:
        conexion.rollback()
        print(f'Error al eliminar: {error}. Cambios revertidos.')
        return 0
 
    finally:
        cursor.close()
        conexion.close()

Borrado lógico (soft delete)

El borrado lógico es un patrón ampliamente utilizado que consiste en marcar registros como eliminados sin borrarlos físicamente de la base de datos:

class ProductoRepository:
    def __init__(self, conexion):
        self.conexion = conexion
 
    def soft_delete(self, producto_id):
        cursor = self.conexion.cursor()
        cursor.execute(
            'UPDATE productos SET activo = 0, eliminado_en = NOW() '
            'WHERE id = %s AND activo = 1',
            (producto_id,)
        )
        self.conexion.commit()
 
        if cursor.rowcount == 0:
            return {'eliminado': False, 'mensaje': 'Producto no encontrado o ya eliminado'}
        return {'eliminado': True}
 
    def restaurar(self, producto_id):
        cursor = self.conexion.cursor()
        cursor.execute(
            'UPDATE productos SET activo = 1, eliminado_en = NULL '
            'WHERE id = %s AND activo = 0',
            (producto_id,)
        )
        self.conexion.commit()
 
        if cursor.rowcount == 0:
            return {'restaurado': False, 'mensaje': 'Producto no encontrado o ya activo'}
        return {'restaurado': True}
 
    def listar_activos(self):
        cursor = self.conexion.cursor(dictionary=True)
        cursor.execute('SELECT id, nombre, precio FROM productos WHERE activo = 1 ORDER BY nombre')
        return cursor.fetchall()
 
    def listar_papelera(self):
        cursor = self.conexion.cursor(dictionary=True)
        cursor.execute(
            'SELECT id, nombre, eliminado_en FROM productos '
            'WHERE activo = 0 ORDER BY eliminado_en DESC'
        )
        return cursor.fetchall()

Caso práctico

Veamos un sistema completo de gestión de eliminaciones con auditoría:

import mysql.connector
from datetime import datetime
 
def purgar_productos_antiguos(dias_limite=90):
    """
    Elimina permanentemente productos que fueron marcados como eliminados
    hace más de X días. Registra la operación en el log de auditoría.
    """
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda',
        autocommit=False
    )
 
    try:
        cursor = conexion.cursor(dictionary=True)
 
        # Encontrar productos para purgar
        cursor.execute(
            'SELECT id, nombre, eliminado_en FROM productos '
            'WHERE activo = 0 AND eliminado_en < DATE_SUB(NOW(), INTERVAL %s DAY)',
            (dias_limite,)
        )
        productos_a_purgar = cursor.fetchall()
 
        if not productos_a_purgar:
            print('No hay productos para purgar')
            return 0
 
        print(f'Productos a purgar ({len(productos_a_purgar)}):')
        for p in productos_a_purgar:
            print(f"  [{p['id']}] {p['nombre']} (eliminado: {p['eliminado_en']})")
 
        # Registrar en auditoría antes de eliminar
        for p in productos_a_purgar:
            cursor.execute(
                'INSERT INTO audit_log (tabla, registro_id, accion, detalles, fecha) '
                'VALUES (%s, %s, %s, %s, NOW())',
                ('productos', p['id'], 'PURGE', f"Purgado después de {dias_limite} días")
            )
 
        # Eliminar permanentemente
        ids = [p['id'] for p in productos_a_purgar]
        placeholders = ', '.join(['%s'] * len(ids))
        cursor.execute(f'DELETE FROM productos WHERE id IN ({placeholders})', tuple(ids))
 
        conexion.commit()
        print(f'\n{cursor.rowcount} productos purgados permanentemente')
        return cursor.rowcount
 
    except Exception as error:
        conexion.rollback()
        print(f'Error durante la purga: {error}')
        return 0
 
    finally:
        cursor.close()
        conexion.close()

Manejo de errores

Los errores más comunes al eliminar están relacionados con las restricciones de claves foráneas:

from mysql.connector import Error, errorcode
 
def eliminar_seguro(conexion, tabla, registro_id):
    try:
        cursor = conexion.cursor()
        cursor.execute(f'DELETE FROM {tabla} WHERE id = %s', (registro_id,))
        conexion.commit()
 
        if cursor.rowcount == 0:
            return {'error': 'NOT_FOUND', 'mensaje': 'Registro no encontrado'}
 
        return {'exito': True, 'eliminados': cursor.rowcount}
 
    except Error as error:
        conexion.rollback()
        if error.errno == errorcode.ER_ROW_IS_REFERENCED_2:
            return {
                'error': 'REFERENCIA',
                'mensaje': 'No se puede eliminar: otros registros dependen de este'
            }
        elif error.errno == errorcode.ER_LOCK_WAIT_TIMEOUT:
            return {
                'error': 'TIMEOUT',
                'mensaje': 'El registro está bloqueado por otra operación'
            }
        elif error.errno == errorcode.ER_LOCK_DEADLOCK:
            return {
                'error': 'DEADLOCK',
                'mensaje': 'Se detectó un interbloqueo. Intenta de nuevo'
            }
        else:
            return {'error': 'INTERNO', 'mensaje': f'Error: {error.msg}'}
 
    finally:
        cursor.close()

En la siguiente sección veremos cómo conectar MySQL con PHP, uno de los lenguajes más utilizados en desarrollo web.

Escrito por Eduardo Lázaro