Crear una API RESTful con Flask y MySQL

Crear una API REST con Flask* y MySQL es una excelente manera de gestionar datos de manera persistente en una base de datos relacional. A continuación, te guiaré paso a paso para construir una API que permita gestionar una lista de tareas (todo list) utilizando Flask y MySQL.

Prerrequisitos

Antes de comenzar, necesitas tener instalado:

  • Python 3.x
  • pip (gestor de paquetes de Python)
  • MySQL Server

Paso 1: Configuración del entorno

  1. Crea un nuevo directorio para tu proyecto:
bash 
mkdir flask-mysql-api
cd flask-mysql-api
  1. Crea un entorno virtual:
bash 
python -m venv venv
  1. Activa el entorno virtual:
  • En Windows:
bash 
venvScriptsactivate
  • En macOS/Linux:
bash 
source venv/bin/activate
  1. Instala las dependencias necesarias:
bash 
pip install flask
pip install flask-cors
pip install mysql-connector-python

Paso 2: Configurar la base de datos MySQL

  1. Crea la base de datos y la tabla en MySQL:
sql 
CREATE DATABASE todo_app;
USE todo_app;

CREATE TABLE todos (
  id INT AUTO_INCREMENT PRIMARY KEY,
  title VARCHAR(255) NOT NULL,
  completed BOOLEAN DEFAULT FALSE,
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Paso 3: Crea un archivo db.py para manejar la conexión a la base de datos:

python 
# db.py
import mysql.connector

def get_db_connection():
  try:
      # connection = mysql.connector.connect(
      connection = mysql.connector.connect(
          host="localhost",
          user="root",
          passwd="",
          database="todo_app",
          charset='utf8mb4',
          collation='utf8mb4_unicode_ci',
          raise_on_warnings=True

      )
      if connection.is_connected():
          # print("Conexión exitosa a la BD")
          return connection

  except mysql.connector.Error as error:
      print(f"No se pudo conectar: {error}")

Crea el archivo principal app.py:

python 
from flask import Flask, request, jsonify
from flask_cors import CORS
from db import get_db_connection
import mysql.connector

app = Flask(__name__)
CORS(app)

@app.route('/')
def home():
  return jsonify({"message": "Bienvenido a la API de Tareas con MySQL"})

# Implementación de endpoints CRUD aquí...

if __name__ == '__main__':
  app.run(debug=True)

Paso 4: Implementar los endpoints CRUD

Añade los siguientes endpoints a app.py:

python 
# Crear una nueva tarea (CREATE)
@app.route('/todos', methods=['POST'])
def create_todo():
  data = request.get_json()
  
  if 'title' not in data:
      return jsonify({"error": "El título es requerido"}), 400
  
  connection = get_db_connection()
  if not connection:
      return jsonify({"error": "Error de conexión a la base de datos"}), 500
  
  cursor = connection.cursor(dictionary=True)
  try:
      sql = "INSERT INTO todos (title) VALUES (%s)"
      cursor.execute(sql, (data['title'],))
      connection.commit()
      
      # Obtener la tarea recién creada
      sql = "SELECT * FROM todos WHERE id = %s"
      cursor.execute(sql, (cursor.lastrowid,))
      new_todo = cursor.fetchone()
      
      return jsonify(new_todo), 201
  except mysql.connector.Error as err:
      return jsonify({"error": str(err)}), 500
  finally:
      cursor.close()
      connection.close()

# Obtener todas las tareas (READ)
@app.route('/todos', methods=['GET'])
def get_todos():
  connection = get_db_connection()
  if not connection:
      return jsonify({"error": "Error de conexión a la base de datos"}), 500
  
  cursor = connection.cursor(dictionary=True)
  try:
      cursor.execute("SELECT * FROM todos ORDER BY created_at DESC")
      todos = cursor.fetchall()
      return jsonify(todos)
  except mysql.connector.Error as err:
      return jsonify({"error": str(err)}), 500
  finally:
      cursor.close()
      connection.close()

# Obtener una tarea específica (READ)
@app.route('/todos/<int:todo_id>', methods=['GET'])
def get_todo(todo_id):
  connection = get_db_connection()
  if not connection:
      return jsonify({"error": "Error de conexión a la base de datos"}), 500
  
  cursor = connection.cursor(dictionary=True)
  try:
      cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
      todo = cursor.fetchone()
      
      if todo is None:
          return jsonify({"error": "Tarea no encontrada"}), 404
          
      return jsonify(todo)
  except mysql.connector.Error as err:
      return jsonify({"error": str(err)}), 500
  finally:
      cursor.close()
      connection.close()

# Actualizar una tarea (UPDATE)
@app.route('/todos/<int:todo_id>', methods=['PUT'])
def update_todo(todo_id):
  data = request.get_json()
  connection = get_db_connection()
  if not connection:
      return jsonify({"error": "Error de conexión a la base de datos"}), 500
  
  cursor = connection.cursor(dictionary=True)
  try:
      # Verificar si la tarea existe
      cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
      todo = cursor.fetchone()
      
      if todo is None:
          return jsonify({"error": "Tarea no encontrada"}), 404
      
      # Actualizar la tarea
      sql = "UPDATE todos SET title = %s, completed = %s WHERE id = %s"
      title = data.get('title', todo['title'])
      completed = data.get('completed', todo['completed'])
      cursor.execute(sql, (title, completed, todo_id))
      connection.commit()
      
      # Obtener la tarea actualizada
      cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
      updated_todo = cursor.fetchone()
      
      return jsonify(updated_todo)
  except mysql.connector.Error as err:
      return jsonify({"error": str(err)}), 500
  finally:
      cursor.close()
      connection.close()

# Eliminar una tarea (DELETE)
@app.route('/todos/<int:todo_id>', methods=['DELETE'])
def delete_todo(todo_id):
  connection = get_db_connection()
  if not connection:
      return jsonify({"error": "Error de conexión a la base de datos"}), 500
  
  cursor = connection.cursor(dictionary=True)
  try:
      # Verificar si la tarea existe
      cursor.execute("SELECT * FROM todos WHERE id = %s", (todo_id,))
      todo = cursor.fetchone()
      
      if todo is None:
          return jsonify({"error": "Tarea no encontrada"}), 404
      
      # Eliminar la tarea
      cursor.execute("DELETE FROM todos WHERE id = %s", (todo_id,))
      connection.commit()
      
      return '', 204
  except mysql.connector.Error as err:
      return jsonify({"error": str(err)}), 500
  finally:
      cursor.close()
      connection.close()

Paso 5: Manejo de errores

Añade manejo de errores específicos para la base de datos:

python 
@app.errorhandler(400)
def bad_request(error):
  return jsonify({"error": "Solicitud incorrecta"}), 400

@app.errorhandler(404)
def not_found(error):
  return jsonify({"error": "Recurso no encontrado"}), 404

@app.errorhandler(500)
def internal_server_error(error):
  return jsonify({"error": "Error interno del servidor"}), 500

Paso 6: Ejecutar la aplicación

  1. Asegúrate de que MySQL esté corriendo
  2. Ejecuta la aplicación:
bash 
python app.py

Probando la API

Crear una nueva tarea:

bash 
curl -X POST -H "Content-Type: application/json" -d '{"title":"Comprar leche"}' http://127.0.0.1:5000/todos

Obtener todas las tareas:

bash 
curl http://127.0.0.1:5000/todos

Obtener una tarea específica:

bash 
curl http://127.0.0.1:5000/todos/1

Actualizar una tarea:

bash 
curl -X PUT -H "Content-Type: application/json" -d '{"completed":true}' http://127.0.0.1:5000/todos/1

Eliminar una tarea:

bash 
curl -X DELETE http://127.0.0.1:5000/todos/1