Cómo Conectar Python con MySQL: Guía paso a paso con ejemplos

Python es el lenguaje favorito para ciencia de datos, automatización y desarrollo backend. MySQL es una de las bases de datos relacionales más usadas del mundo. Juntos forman una combinación que aparece en miles de proyectos reales: desde pequeñas aplicaciones web hasta sistemas de análisis de datos empresariales.

En este artículo vas a aprender a conectar Python con MySQL desde cero: instalar el conector, establecer la conexión, ejecutar operaciones CRUD completas, usar consultas parametrizadas de forma segura y manejar transacciones. Con código real y funcional en cada paso.

Si todavía no tienes claros los fundamentos de SQL, te recomendamos leer primero nuestra guía sobre qué es SQL y para qué sirve y los comandos SQL básicos antes de continuar. Y si quieres tener MySQL disponible sin instalar nada, puedes usar Docker para levantar MySQL en segundos.

Las librerías disponibles para conectar Python con MySQL

Antes de escribir código, necesitas elegir qué librería usarás para la conexión. Las principales opciones son:

En este artículo usamos mysql-connector-python (la oficial) para el CRUD y las operaciones básicas, y SQLAlchemy para mostrar cómo se hace con un ORM. Ambas son las más usadas en proyectos profesionales.

Paso 1: Preparar el entorno

Instalar MySQL

Necesitas MySQL corriendo en tu máquina. Tienes tres opciones:

# Opción A: MySQL con Docker (la más rápida, sin instalación)
docker run -d \
  --name mysql-dev \
  -e MYSQL_ROOT_PASSWORD=secreto \
  -e MYSQL_DATABASE=tienda \
  -e MYSQL_USER=usuario \
  -e MYSQL_PASSWORD=clave123 \
  -p 3306:3306 \
  mysql:8.0

# Opción B: Instalación en Ubuntu/Debian
sudo apt update && sudo apt install mysql-server -y
sudo mysql_secure_installation

# Opción C: Descarga el instalador de mysql.com
# → https://dev.mysql.com/downloads/mysql/

Instalar mysql-connector-python

# Instalar el conector oficial
pip install mysql-connector-python

# O PyMySQL si lo prefieres
pip install pymysql

# O SQLAlchemy (incluye soporte para múltiples bases de datos)
pip install sqlalchemy pymysql

# Verificar la instalación
python -c "import mysql.connector; print(mysql.connector.__version__)"

Crear la base de datos y tabla de ejemplo

-- Ejecuta esto en tu cliente MySQL (MySQL Workbench, DBeaver, terminal...)

CREATE DATABASE IF NOT EXISTS tienda
CHARACTER SET utf8mb4
COLLATE utf8mb4_unicode_ci;

USE tienda;

CREATE TABLE IF NOT EXISTS productos (
    id          INT AUTO_INCREMENT PRIMARY KEY,
    nombre      VARCHAR(100) NOT NULL,
    precio      DECIMAL(10,2) NOT NULL,
    stock       INT DEFAULT 0,
    categoria   VARCHAR(50),
    activo      BOOLEAN DEFAULT TRUE,
    creado_en   TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    actualizado TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

-- Datos de prueba
INSERT INTO productos (nombre, precio, stock, categoria) VALUES
('Laptop Pro 15',    899.99, 20,  'tecnologia'),
('Mouse Inalámbrico', 29.99, 150, 'periféricos'),
('Teclado Mecánico',  79.99, 80,  'periféricos'),
('Monitor 4K',       399.99, 25,  'tecnologia'),
('Webcam HD',         49.99, 60,  'periféricos');

Paso 2: Establecer la conexión

import mysql.connector
from mysql.connector import Error

# ============================================
# CONEXIÓN BÁSICA
# ============================================

def crear_conexion():
    """Establece y devuelve una conexión a MySQL."""
    try:
        conexion = mysql.connector.connect(
            host="localhost",        # dirección del servidor MySQL
            port=3306,               # puerto (3306 por defecto)
            user="usuario",          # nombre de usuario
            password="clave123",     # contraseña
            database="tienda",       # base de datos a usar
            charset="utf8mb4",       # codificación de caracteres
            use_unicode=True
        )

        if conexion.is_connected():
            info = conexion.get_server_info()
            print(f"✅ Conectado a MySQL Server versión {info}")
            return conexion

    except Error as e:
        print(f"❌ Error al conectar: {e}")
        return None

# Usar la función
conexion = crear_conexion()

# Siempre cerrar la conexión cuando termines
if conexion and conexion.is_connected():
    conexion.close()
    print("Conexión cerrada")

Usar variables de entorno — la forma correcta y segura

Nunca escribas las credenciales directamente en el código. Si subes el archivo a GitHub, quedarán expuestas públicamente.

# .env (NUNCA subir a git — añade .env a tu .gitignore)
DB_HOST=localhost
DB_PORT=3306
DB_USER=usuario
DB_PASSWORD=clave123
DB_NAME=tienda

import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv  # pip install python-dotenv

load_dotenv()  # carga las variables del archivo .env

def crear_conexion():
    """Conexión segura usando variables de entorno."""
    try:
        conexion = mysql.connector.connect(
            host     = os.getenv("DB_HOST", "localhost"),
            port     = int(os.getenv("DB_PORT", 3306)),
            user     = os.getenv("DB_USER"),
            password = os.getenv("DB_PASSWORD"),
            database = os.getenv("DB_NAME"),
            charset  = "utf8mb4"
        )
        return conexion

    except Error as e:
        print(f"❌ Error de conexión: {e}")
        return None

Paso 3: El cursor — el ejecutor de consultas

El cursor es el objeto que ejecuta las consultas SQL y gestiona los resultados. Es tu interfaz directa con la base de datos.

conexion = crear_conexion()

if conexion:
    # Crear un cursor
    cursor = conexion.cursor()

    # Cursor con diccionarios (devuelve filas como dicts en lugar de tuplas)
    cursor_dict = conexion.cursor(dictionary=True)

    # Con cursor normal: accedes por índice
    cursor.execute("SELECT id, nombre, precio FROM productos LIMIT 1")
    fila = cursor.fetchone()
    print(fila[0])      # id: 1
    print(fila[1])      # nombre: "Laptop Pro 15"

    # Con cursor dictionary: accedes por nombre de columna
    cursor_dict.execute("SELECT id, nombre, precio FROM productos LIMIT 1")
    fila = cursor_dict.fetchone()
    print(fila["id"])       # 1
    print(fila["nombre"])   # "Laptop Pro 15"
    print(fila["precio"])   # 899.99

    cursor.close()
    cursor_dict.close()
    conexion.close()

Paso 4: CRUD completo — leer, crear, actualizar y eliminar

📖 READ — leer datos

import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv

load_dotenv()

def obtener_conexion():
    return mysql.connector.connect(
        host=os.getenv("DB_HOST"), user=os.getenv("DB_USER"),
        password=os.getenv("DB_PASSWORD"), database=os.getenv("DB_NAME"),
        charset="utf8mb4"
    )

# ── Obtener todos los productos ──────────────────────────────
def obtener_todos_los_productos():
    conexion = obtener_conexion()
    cursor = conexion.cursor(dictionary=True)

    try:
        cursor.execute("""
            SELECT id, nombre, precio, stock, categoria
            FROM productos
            WHERE activo = TRUE
            ORDER BY nombre
        """)
        productos = cursor.fetchall()  # devuelve lista de dicts
        return productos

    except Error as e:
        print(f"Error al leer: {e}")
        return []

    finally:
        cursor.close()
        conexion.close()

# ── Obtener un producto por ID ───────────────────────────────
def obtener_producto_por_id(producto_id):
    conexion = obtener_conexion()
    cursor = conexion.cursor(dictionary=True)

    try:
        # ⚠️ SIEMPRE usa consultas parametrizadas — nunca f-strings con datos externos
        cursor.execute(
            "SELECT * FROM productos WHERE id = %s AND activo = TRUE",
            (producto_id,)  # la coma es necesaria para que sea tupla
        )
        return cursor.fetchone()  # devuelve un dict o None

    except Error as e:
        print(f"Error: {e}")
        return None

    finally:
        cursor.close()
        conexion.close()

# ── Obtener con filtros y paginación ────────────────────────
def buscar_productos(categoria=None, precio_max=None, pagina=1, por_pagina=10):
    conexion = obtener_conexion()
    cursor = conexion.cursor(dictionary=True)

    try:
        query = "SELECT id, nombre, precio, stock FROM productos WHERE activo = TRUE"
        params = []

        if categoria:
            query += " AND categoria = %s"
            params.append(categoria)

        if precio_max:
            query += " AND precio <= %s"
            params.append(precio_max)

        query += " ORDER BY nombre LIMIT %s OFFSET %s"
        params.append(por_pagina)
        params.append((pagina - 1) * por_pagina)

        cursor.execute(query, tuple(params))
        return cursor.fetchall()

    finally:
        cursor.close()
        conexion.close()

# Probar las funciones
productos = obtener_todos_los_productos()
print(f"Total productos: {len(productos)}")
for p in productos:
    print(f"  [{p['id']}] {p['nombre']} — ${p['precio']}")

producto = obtener_producto_por_id(1)
if producto:
    print(f"\nProducto encontrado: {producto['nombre']}")

tecnologia = buscar_productos(categoria="tecnologia", precio_max=500)
print(f"\nTecnología < $500: {len(tecnologia)} productos")

➕ CREATE — insertar datos

# ── Insertar un producto ─────────────────────────────────────
def insertar_producto(nombre, precio, stock, categoria):
    conexion = obtener_conexion()
    cursor = conexion.cursor()

    try:
        sql = """
            INSERT INTO productos (nombre, precio, stock, categoria)
            VALUES (%s, %s, %s, %s)
        """
        valores = (nombre, precio, stock, categoria)
        cursor.execute(sql, valores)
        conexion.commit()  # ← IMPORTANTE: confirmar la transacción

        nuevo_id = cursor.lastrowid  # ID del registro insertado
        print(f"✅ Producto creado con ID: {nuevo_id}")
        return nuevo_id

    except Error as e:
        conexion.rollback()  # revertir si hay error
        print(f"❌ Error al insertar: {e}")
        return None

    finally:
        cursor.close()
        conexion.close()

# ── Insertar múltiples productos a la vez ────────────────────
def insertar_varios_productos(lista_productos):
    """
    lista_productos: lista de tuplas (nombre, precio, stock, categoria)
    """
    conexion = obtener_conexion()
    cursor = conexion.cursor()

    try:
        sql = """
            INSERT INTO productos (nombre, precio, stock, categoria)
            VALUES (%s, %s, %s, %s)
        """
        cursor.executemany(sql, lista_productos)  # ← mucho más eficiente que un loop
        conexion.commit()

        print(f"✅ {cursor.rowcount} productos insertados")
        return cursor.rowcount

    except Error as e:
        conexion.rollback()
        print(f"❌ Error: {e}")
        return 0

    finally:
        cursor.close()
        conexion.close()

# Probar
nuevo_id = insertar_producto("Auriculares BT", 59.99, 40, "audio")

nuevos = [
    ("SSD 1TB",      89.99,  35, "almacenamiento"),
    ("Cable HDMI",    9.99, 200, "cables"),
    ("Hub USB-C",    29.99,  75, "periféricos"),
]
insertar_varios_productos(nuevos)

✏️ UPDATE — actualizar datos

# ── Actualizar un producto ───────────────────────────────────
def actualizar_producto(producto_id, **campos):
    """
    Actualiza solo los campos especificados.
    Uso: actualizar_producto(1, precio=849.99, stock=18)
    """
    if not campos:
        print("No se especificaron campos a actualizar")
        return False

    conexion = obtener_conexion()
    cursor = conexion.cursor()

    try:
        # Construir la query dinámicamente con los campos recibidos
        sets = ", ".join([f"{campo} = %s" for campo in campos.keys()])
        valores = list(campos.values()) + [producto_id]

        sql = f"UPDATE productos SET {sets} WHERE id = %s"
        cursor.execute(sql, valores)
        conexion.commit()

        if cursor.rowcount > 0:
            print(f"✅ Producto {producto_id} actualizado ({cursor.rowcount} fila/s)")
            return True
        else:
            print(f"⚠️ Producto {producto_id} no encontrado")
            return False

    except Error as e:
        conexion.rollback()
        print(f"❌ Error al actualizar: {e}")
        return False

    finally:
        cursor.close()
        conexion.close()

# ── Actualizar precio de múltiples productos ─────────────────
def aplicar_descuento_categoria(categoria, porcentaje):
    """Aplica un descuento de X% a todos los productos de una categoría."""
    conexion = obtener_conexion()
    cursor = conexion.cursor()

    try:
        sql = """
            UPDATE productos
            SET precio = ROUND(precio * (1 - %s / 100), 2)
            WHERE categoria = %s AND activo = TRUE
        """
        cursor.execute(sql, (porcentaje, categoria))
        conexion.commit()

        print(f"✅ Descuento del {porcentaje}% aplicado a {cursor.rowcount} productos en '{categoria}'")
        return cursor.rowcount

    except Error as e:
        conexion.rollback()
        print(f"❌ Error: {e}")
        return 0

    finally:
        cursor.close()
        conexion.close()

# Probar
actualizar_producto(1, precio=849.99, stock=18)
aplicar_descuento_categoria("periféricos", 10)

🗑️ DELETE — eliminar datos

# ── Borrado lógico (soft delete) — recomendado en producción ──
def desactivar_producto(producto_id):
    """
    No borra físicamente — marca como inactivo.
    Los registros históricos (pedidos) siguen funcionando.
    """
    return actualizar_producto(producto_id, activo=False)

# ── Borrado físico — usar con precaución ─────────────────────
def eliminar_producto(producto_id):
    conexion = obtener_conexion()
    cursor = conexion.cursor()

    try:
        cursor.execute(
            "DELETE FROM productos WHERE id = %s",
            (producto_id,)
        )
        conexion.commit()

        if cursor.rowcount > 0:
            print(f"✅ Producto {producto_id} eliminado")
            return True
        else:
            print(f"⚠️ Producto {producto_id} no encontrado")
            return False

    except Error as e:
        conexion.rollback()
        print(f"❌ Error al eliminar: {e}")
        return False

    finally:
        cursor.close()
        conexion.close()

# Probar borrado lógico
desactivar_producto(6)   # "elimina" sin borrar el registro

Paso 5: Consultas parametrizadas — la seguridad que no es opcional

Este es uno de los temas más críticos. La inyección SQL es el ataque más común contra bases de datos. Ocurre cuando un atacante introduce código SQL malicioso en los datos de entrada. La forma de prevenirla es siempre usar consultas parametrizadas.

# ❌ PELIGROSO: concatenación directa — permite inyección SQL
nombre_busqueda = "' OR '1'='1"  # entrada maliciosa de un usuario

query = f"SELECT * FROM productos WHERE nombre = '{nombre_busqueda}'"
# La query resultante sería:
# SELECT * FROM productos WHERE nombre = '' OR '1'='1'
# → Devuelve TODOS los productos ignorando el filtro ← inyección SQL

# ❌ Igualmente peligroso con format()
query = "SELECT * FROM productos WHERE nombre = '%s'" % nombre_busqueda

# ────────────────────────────────────────────────────────────────

# ✅ SEGURO: consultas parametrizadas — el conector escapa los valores
cursor.execute(
    "SELECT * FROM productos WHERE nombre = %s",
    (nombre_busqueda,)  # el conector escapa automáticamente el valor
)
# La query real que ejecuta MySQL está escapada y es segura
# El valor nunca se interpreta como SQL

# ✅ Con múltiples parámetros
cursor.execute(
    "SELECT * FROM productos WHERE categoria = %s AND precio <= %s AND activo = %s",
    ("tecnologia", 500, True)
)

# ⚠️ Nota importante:
# En mysql-connector-python el placeholder es %s para TODOS los tipos
# (strings, números, booleanos, fechas...)
# No uses ? como en SQLite — en MySQL/connector es siempre %s

Paso 6: Transacciones — operaciones atómicas

Ya vimos en el artículo sobre transacciones en SQL qué son las propiedades ACID. Desde Python puedes controlar transacciones explícitamente:

def procesar_venta(producto_id, cantidad, cliente_id):
    """
    Operación de venta que afecta múltiples tablas.
    Todo se confirma junto o todo se revierte.
    """
    conexion = obtener_conexion()
    cursor = conexion.cursor(dictionary=True)

    try:
        # Desactivar autocommit para control manual
        conexion.autocommit = False

        # 1. Verificar stock disponible (con bloqueo FOR UPDATE)
        cursor.execute(
            "SELECT stock, precio FROM productos WHERE id = %s FOR UPDATE",
            (producto_id,)
        )
        producto = cursor.fetchone()

        if not producto:
            raise ValueError(f"Producto {producto_id} no existe")

        if producto["stock"] < cantidad:
            raise ValueError(
                f"Stock insuficiente. Disponible: {producto['stock']}, solicitado: {cantidad}"
            )

        total = producto["precio"] * cantidad

        # 2. Reducir el stock
        cursor.execute(
            "UPDATE productos SET stock = stock - %s WHERE id = %s",
            (cantidad, producto_id)
        )

        # 3. Registrar la venta
        cursor.execute(
            """INSERT INTO ventas (cliente_id, producto_id, cantidad, total, fecha)
               VALUES (%s, %s, %s, %s, NOW())""",
            (cliente_id, producto_id, cantidad, total)
        )
        venta_id = cursor.lastrowid

        # 4. Si todo fue bien → confirmar
        conexion.commit()
        print(f"✅ Venta #{venta_id} procesada. Total: ${total:.2f}")
        return venta_id

    except ValueError as e:
        conexion.rollback()
        print(f"❌ Error de negocio: {e}")
        return None

    except Error as e:
        conexion.rollback()
        print(f"❌ Error de base de datos: {e}")
        return None

    finally:
        cursor.close()
        conexion.close()

Paso 7: Connection Pool — rendimiento en producción

Abrir y cerrar una conexión a MySQL en cada operación es costoso. En aplicaciones con muchos usuarios simultáneos, el pool de conexiones mantiene un conjunto de conexiones abiertas y listas para reutilizar.

from mysql.connector import pooling

# Crear el pool al iniciar la aplicación (una sola vez)
pool = pooling.MySQLConnectionPool(
    pool_name    = "mi_pool",
    pool_size    = 10,             # máximo 10 conexiones simultáneas
    host         = os.getenv("DB_HOST"),
    user         = os.getenv("DB_USER"),
    password     = os.getenv("DB_PASSWORD"),
    database     = os.getenv("DB_NAME"),
    charset      = "utf8mb4"
)

def obtener_conexion_pool():
    """Obtiene una conexión del pool (más eficiente que crear una nueva)."""
    return pool.get_connection()

# Usar exactamente igual que una conexión normal
def obtener_productos_con_pool():
    conexion = obtener_conexion_pool()  # ← del pool
    cursor = conexion.cursor(dictionary=True)

    try:
        cursor.execute("SELECT id, nombre, precio FROM productos WHERE activo = TRUE")
        return cursor.fetchall()

    finally:
        cursor.close()
        conexion.close()  # ← devuelve la conexión al pool, no la cierra realmente

Paso 8: SQLAlchemy — la forma ORM

Para proyectos más grandes, muchos developers prefieren usar un ORM (Object-Relational Mapper) como SQLAlchemy. Permite trabajar con la base de datos usando clases y objetos Python en lugar de SQL crudo, de la misma forma que usarías Programación Orientada a Objetos.

# pip install sqlalchemy pymysql

from sqlalchemy import create_engine, Column, Integer, String, Numeric, Boolean, DateTime
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.sql import func
import os
from dotenv import load_dotenv

load_dotenv()

# ── Configurar la conexión ───────────────────────────────────
# Formato: dialect+driver://user:pass@host:port/database
DATABASE_URL = (
    f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT', 3306)}/{os.getenv('DB_NAME')}"
    f"?charset=utf8mb4"
)

engine = create_engine(DATABASE_URL, echo=False, pool_size=10)
Base = declarative_base()
Session = sessionmaker(bind=engine)

# ── Definir el modelo (equivale a la tabla) ──────────────────
class Producto(Base):
    __tablename__ = "productos"

    id        = Column(Integer, primary_key=True, autoincrement=True)
    nombre    = Column(String(100), nullable=False)
    precio    = Column(Numeric(10, 2), nullable=False)
    stock     = Column(Integer, default=0)
    categoria = Column(String(50))
    activo    = Column(Boolean, default=True)
    creado_en = Column(DateTime, server_default=func.now())

    def __repr__(self):
        return f""

# ── CRUD con SQLAlchemy ──────────────────────────────────────
def demo_sqlalchemy():
    session = Session()

    try:
        # CREATE — insertar
        nuevo = Producto(nombre="Parlante BT", precio=39.99, stock=50, categoria="audio")
        session.add(nuevo)
        session.commit()
        print(f"✅ Insertado: {nuevo}")

        # READ — leer
        todos = session.query(Producto).filter(Producto.activo == True).all()
        print(f"\nTotal activos: {len(todos)}")

        # Filtros avanzados
        tecnologia = (session.query(Producto)
                      .filter(Producto.categoria == "tecnologia")
                      .filter(Producto.precio < 500)
                      .order_by(Producto.precio)
                      .all())

        # UPDATE — actualizar
        producto = session.query(Producto).filter(Producto.id == 1).first()
        if producto:
            producto.precio = 849.99
            producto.stock  = 18
            session.commit()
            print(f"\n✅ Actualizado: {producto}")

        # DELETE — borrado lógico
        producto_borrar = session.query(Producto).filter(Producto.id == nuevo.id).first()
        if producto_borrar:
            producto_borrar.activo = False
            session.commit()
            print(f"✅ Desactivado: {producto_borrar.nombre}")

    except Exception as e:
        session.rollback()
        print(f"❌ Error: {e}")

    finally:
        session.close()

demo_sqlalchemy()

Errores comunes y cómo evitarlos

❌ Olvidar el commit() en INSERT, UPDATE y DELETE

# ❌ Sin commit: los cambios existen en la transacción pero no se guardan en disco
cursor.execute("INSERT INTO productos (nombre, precio) VALUES ('Producto', 9.99)")
# Si cierras la conexión aquí, el INSERT se pierde

# ✅ Siempre commit() después de operaciones de escritura
cursor.execute("INSERT INTO productos (nombre, precio) VALUES (%s, %s)", ("Producto", 9.99))
conexion.commit()  # ← imprescindible

❌ No cerrar cursores y conexiones

# ❌ Pérdida de recursos — las conexiones se agotan
def mala_practica():
    conexion = obtener_conexion()
    cursor = conexion.cursor()
    cursor.execute("SELECT * FROM productos")
    return cursor.fetchall()
    # cursor y conexion nunca se cierran → memory leak

# ✅ Siempre usar try/finally o context managers
def buena_practica():
    conexion = obtener_conexion()
    cursor = conexion.cursor(dictionary=True)
    try:
        cursor.execute("SELECT * FROM productos")
        return cursor.fetchall()
    finally:
        cursor.close()
        conexion.close()

❌ Usar fetchall() en consultas que devuelven millones de filas

# ❌ fetchall() carga TODOS los resultados en memoria
cursor.execute("SELECT * FROM logs")  # tabla con millones de filas
todos = cursor.fetchall()  # puede agotar la memoria del servidor

# ✅ Usar fetchmany() o iterar con el cursor directamente
cursor.execute("SELECT * FROM logs")
while True:
    filas = cursor.fetchmany(1000)  # procesa 1000 a la vez
    if not filas:
        break
    for fila in filas:
        procesar(fila)

❌ Concatenar variables en queries (SQL Injection)

# ❌ NUNCA — inyección SQL garantizada
id_usuario = request.args.get("id")
cursor.execute(f"SELECT * FROM usuarios WHERE id = {id_usuario}")

# ✅ SIEMPRE — consultas parametrizadas
cursor.execute("SELECT * FROM usuarios WHERE id = %s", (id_usuario,))

Resumen: lo que aprendiste hoy

• ✅ mysql-connector-python es el conector oficial — instálalo con pip install mysql-connector-python
• ✅ Siempre guarda las credenciales en variables de entorno (.env), nunca en el código
• ✅ El cursor ejecuta las consultas — usa dictionary=True para acceder por nombre de columna
• ✅ CRUD completo: fetchall() / fetchone() para leer, execute() + commit() para escribir
• ✅ Las consultas parametrizadas con %s previenen la inyección SQL — siempre obligatorias
• ✅ executemany() inserta múltiples filas de forma mucho más eficiente que un bucle
• ✅ Las transacciones con commit() y rollback() garantizan la integridad de los datos
• ✅ El connection pool mejora el rendimiento en producción reutilizando conexiones
• ✅ SQLAlchemy permite trabajar con MySQL usando clases Python sin escribir SQL directamente
• ✅ Siempre cierra cursores y conexiones en un bloque finally o usa context managers

🧪 ¿Tienes los fundamentos de Python y SQL bien sólidos?

Conectar Python con MySQL requiere dominar los fundamentos de ambos: bucles, funciones y manejo de errores en Python, más SELECT, INSERT, UPDATE y DELETE en SQL. Comprueba dónde estás:

👉 Test: Python Básico 👉 Test: SQL Básico 

¿Lograste conectar Python con MySQL y ejecutar tus primeras consultas? ¿Usas mysql-connector-python o prefieres SQLAlchemy? ¿Tuviste algún error en el camino? Cuéntanos en los comentarios 👇 — respondemos todos. 🚀