Conectar Python con MySQL

Python es uno de los lenguajes más utilizados para el análisis de datos, desarrollo web y automatización, y MySQL es una de las bases de datos más populares del mundo. Conectar ambas tecnologías te permite construir aplicaciones potentes que interactúan con datos almacenados de forma estructurada. En este artículo aprenderás a establecer la conexión usando el conector oficial mysql-connector-python, gestionar pools de conexiones y manejar errores de forma apropiada.

Requisitos previos

Necesitas tener Python 3.7 o superior instalado en tu sistema. Puedes verificar tu versión ejecutando python --version o python3 --version en la terminal. También necesitas un servidor MySQL en ejecución con un usuario que tenga permisos para crear bases de datos y tablas.

Instalación

Instala el conector oficial de MySQL para Python usando pip:

# pip install mysql-connector-python

Este paquete es mantenido por Oracle, el mismo equipo detrás de MySQL, lo que garantiza compatibilidad y actualizaciones constantes. Existe una alternativa popular llamada PyMySQL que es una implementación pura de Python, pero el conector oficial ofrece mejor rendimiento gracias a su extensión en C.

Código completo

A continuación se muestra un ejemplo completo que establece una conexión, ejecuta una consulta de prueba y cierra la conexión:

import mysql.connector
 
def conectar():
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda'
    )
 
    cursor = conexion.cursor()
    cursor.execute('SELECT 1 + 1 AS resultado')
    fila = cursor.fetchone()
    print(f'Conexión exitosa. Resultado: {fila[0]}')
 
    cursor.close()
    conexion.close()
 
conectar()

Salida esperada:

Conexión exitosa. Resultado: 2

Explicación paso a paso

La función mysql.connector.connect() crea una conexión al servidor MySQL. Acepta múltiples parámetros de configuración que controlan el comportamiento de la conexión:

import mysql.connector
 
config = {
    'host': 'localhost',         # Dirección del servidor
    'port': 3306,                # Puerto (3306 por defecto)
    'user': 'root',              # Usuario de MySQL
    'password': 'tu_contraseña', # Contraseña
    'database': 'tienda',        # Base de datos
    'charset': 'utf8mb4',        # Juego de caracteres
    'collation': 'utf8mb4_general_ci',
    'autocommit': False,         # Control manual de transacciones
    'connection_timeout': 10,    # Timeout en segundos
    'use_pure': False,           # Usar extensión C para mejor rendimiento
    'raise_on_warnings': True    # Lanzar excepción en warnings
}
 
conexion = mysql.connector.connect(**config)
print(f'Conectado a MySQL Server versión {conexion.get_server_info()}')
print(f'ID de conexión: {conexion.connection_id}')
conexion.close()

Salida esperada:

Conectado a MySQL Server versión 8.0.35
ID de conexión: 42

Pasar la configuración como diccionario es una práctica recomendada porque permite cargar los valores desde un archivo de configuración o variables de entorno sin modificar el código.

Usar el context manager (with)

Python ofrece el patrón with para gestionar recursos que necesitan ser liberados al terminar. Aunque mysql.connector.connect() no implementa el protocolo de context manager por defecto, puedes crear uno fácilmente:

import mysql.connector
from contextlib import contextmanager
 
@contextmanager
def obtener_conexion():
    conexion = mysql.connector.connect(
        host='localhost',
        user='root',
        password='tu_contraseña',
        database='tienda'
    )
    try:
        yield conexion
    finally:
        conexion.close()
 
# Uso con context manager
with obtener_conexion() as conexion:
    cursor = conexion.cursor()
    cursor.execute('SELECT DATABASE()')
    resultado = cursor.fetchone()
    print(f'Base de datos actual: {resultado[0]}')
    cursor.close()
# La conexión se cierra automáticamente al salir del bloque with

Salida esperada:

Base de datos actual: tienda

Pool de conexiones

Para aplicaciones que necesitan manejar múltiples solicitudes concurrentes, como un servidor web, crear y cerrar conexiones constantemente es ineficiente. Un pool de conexiones mantiene varias conexiones abiertas y las reutiliza:

import mysql.connector
from mysql.connector import pooling
 
# Crear el pool (se recomienda hacerlo una sola vez al iniciar la aplicación)
pool = pooling.MySQLConnectionPool(
    pool_name='tienda_pool',
    pool_size=5,                 # Número de conexiones en el pool
    pool_reset_session=True,     # Resetear sesión al devolver al pool
    host='localhost',
    user='root',
    password='tu_contraseña',
    database='tienda',
    charset='utf8mb4',
    autocommit=True
)
 
def consultar_productos():
    conexion = pool.get_connection()
    try:
        cursor = conexion.cursor(dictionary=True)
        cursor.execute('SELECT id, nombre, precio FROM productos LIMIT 5')
        productos = cursor.fetchall()
 
        print(f'Productos encontrados: {len(productos)}')
        for p in productos:
            print(f"  {p['nombre']}: ${p['precio']}")
 
        cursor.close()
    finally:
        conexion.close()  # Devuelve la conexión al pool
 
consultar_productos()

Salida esperada:

Productos encontrados: 5
  Laptop HP Pavilion: $12999.99
  Mouse Logitech MX Master: $1599.00
  Teclado Mecánico Corsair K70: $2299.50
  Monitor Samsung 27": $6499.00
  Auriculares Sony WH-1000XM5: $5999.99

Cuando llamas a conexion.close() en una conexión obtenida del pool, la conexión no se destruye realmente sino que se devuelve al pool para ser reutilizada. Si todas las conexiones del pool están en uso, get_connection() lanzará un error PoolError.

Caso práctico

Veamos cómo estructurar la conexión en una aplicación Flask:

import mysql.connector
from mysql.connector import pooling
import os
 
class BaseDatos:
    _pool = None
 
    @classmethod
    def inicializar(cls):
        cls._pool = pooling.MySQLConnectionPool(
            pool_name='app_pool',
            pool_size=10,
            host=os.getenv('DB_HOST', 'localhost'),
            user=os.getenv('DB_USER', 'root'),
            password=os.getenv('DB_PASSWORD', ''),
            database=os.getenv('DB_NAME', 'tienda'),
            charset='utf8mb4',
            autocommit=True
        )
        print('Pool de conexiones inicializado')
 
    @classmethod
    @contextmanager
    def obtener_conexion(cls):
        conexion = cls._pool.get_connection()
        try:
            yield conexion
        finally:
            conexion.close()
 
    @classmethod
    @contextmanager
    def obtener_cursor(cls, dictionary=True):
        with cls.obtener_conexion() as conexion:
            cursor = conexion.cursor(dictionary=dictionary)
            try:
                yield cursor
            finally:
                cursor.close()
 
# Inicializar al arrancar la aplicación
BaseDatos.inicializar()
 
# Uso en cualquier parte del código
with BaseDatos.obtener_cursor() as cursor:
    cursor.execute('SELECT COUNT(*) AS total FROM productos')
    resultado = cursor.fetchone()
    print(f"Total de productos: {resultado['total']}")

Manejo de errores

La conexión puede fallar por diversas razones. El conector de MySQL proporciona excepciones específicas para cada tipo de error:

import mysql.connector
from mysql.connector import Error, errorcode
 
def conectar_con_manejo():
    try:
        conexion = mysql.connector.connect(
            host='localhost',
            user='root',
            password='contraseña_incorrecta',
            database='base_inexistente'
        )
        conexion.close()
 
    except Error as error:
        if error.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print('Credenciales incorrectas. Verifica usuario y contraseña.')
        elif error.errno == errorcode.ER_BAD_DB_ERROR:
            print('La base de datos especificada no existe.')
        elif error.errno == errorcode.CR_CONN_HOST_ERROR:
            print('No se pudo conectar al servidor MySQL. ¿Está en ejecución?')
        elif error.errno == errorcode.CR_CONNECTION_ERROR:
            print('Error de conexión. Verifica host y puerto.')
        else:
            print(f'Error MySQL [{error.errno}]: {error.msg}')
 
conectar_con_manejo()

También puedes verificar si la conexión sigue activa y reconectar automáticamente:

def ejecutar_consulta(conexion, sql, params=None):
    try:
        if not conexion.is_connected():
            print('Reconectando...')
            conexion.reconnect(attempts=3, delay=2)
 
        cursor = conexion.cursor(dictionary=True)
        cursor.execute(sql, params)
        resultados = cursor.fetchall()
        cursor.close()
        return resultados
 
    except Error as error:
        print(f'Error en la consulta: {error}')
        return []

Ahora que sabes cómo establecer la conexión entre Python y MySQL, en el siguiente artículo aprenderás a ejecutar consultas SELECT para recuperar datos.

Escrito por Eduardo Lázaro