Tutorial Flask: Rutas endpoints REST GET

Estructura básica de un controlador REST

En Flask, un controlador REST es responsable de gestionar las solicitudes HTTP y proporcionar respuestas adecuadas en formato JSON. Para construir un controlador REST básico, es esencial conocer cómo definir rutas, manejar métodos HTTP y estructurar las respuestas.

A continuación se presenta un ejemplo de cómo estructurar un controlador REST utilizando Flask:

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/recursos', methods=['GET'])
def obtener_recursos():
    # Lógica para obtener los recursos
    recursos = [
        {'id': 1, 'nombre': 'Recurso 1'},
        {'id': 2, 'nombre': 'Recurso 2'}
    ]
    return jsonify(recursos), 200

En este ejemplo:

• Se importa Flask, request y jsonify para manejar la aplicación, las solicitudes y las respuestas en formato JSON.
• Se crea una instancia de la aplicación Flask.
• Se utiliza el decorador @app.route para definir una ruta endpoint /api/recursos que responde al método HTTP GET.
• La función obtener_recursos es el controlador que maneja las solicitudes a esa ruta y método.

Es importante destacar que el uso de decoradores en Flask permite asociar rutas a funciones de manera sencilla y clara.

Para manejar diferentes métodos HTTP en un mismo endpoint, se puede especificar una lista de métodos:

@app.route('/api/recursos', methods=['GET', 'POST'])
def gestionar_recursos():
    if request.method == 'GET':
        # Lógica para obtener los recursos
        pass
    elif request.method == 'POST':
        # Lógica para crear un nuevo recurso
        pass

En este caso, la función gestionar_recursos maneja tanto solicitudes GET como POST en el endpoint /api/recursos.

Para una mejor organización y modularidad, es recomendable utilizar Blueprints y estructurar los controladores en módulos separados. Por ejemplo, creando un módulo controlador_recursos.py:

from flask import Blueprint, request, jsonify

recursos_bp = Blueprint('recursos_bp', __name__)

@recursos_bp.route('/api/recursos', methods=['GET'])
def obtener_recursos():
    # Lógica para obtener los recursos
    recursos = [
        {'id': 1, 'nombre': 'Recurso 1'},
        {'id': 2, 'nombre': 'Recurso 2'}
    ]
    return jsonify(recursos), 200

Y luego registrarlo en la aplicación principal:

from flask import Flask
from controlador_recursos import recursos_bp

app = Flask(__name__)
app.register_blueprint(recursos_bp)

Esto permite mantener el código organizado y facilita la gestión de múltiples controladores.

Al definir controladores REST, es fundamental manejar correctamente los códigos de estado HTTP y las excepciones. Por ejemplo, si un recurso no se encuentra:

@app.route('/api/recursos/<int:id>', methods=['GET'])
def obtener_recurso(id):
    # Lógica para obtener un recurso específico
    recurso = obtener_recurso_por_id(id)
    if recurso:
        return jsonify(recurso), 200
    else:
        return jsonify({'mensaje': 'Recurso no encontrado'}), 404

En este ejemplo, se utiliza una ruta con parámetro id para obtener un recurso específico y se maneja la posibilidad de que el recurso no exista, devolviendo un código de estado 404 Not Found.

Además, es recomendable separar la lógica de negocio de la capa de presentación, utilizando modelos y servicios para interactuar con la base de datos MySQL a través de SQLAlchemy, y mantener los controladores enfocados en manejar las solicitudes y respuestas.

Finalmente, para ejecutar la aplicación:

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

Activar el modo debug es útil durante el desarrollo, pero debe deshabilitarse en producción por razones de seguridad.

Respuestas con jsonify

En Flask, la función jsonify es fundamental para enviar respuestas en formato JSON desde nuestros endpoints REST. Esta función simplifica la conversión de estructuras de datos de Python en respuestas HTTP JSON serializadas adecuadamente, facilitando el desarrollo de APIs.

Por ejemplo, al responder a una petición que devuelve una lista de usuarios:

from flask import jsonify

@app.route('/api/usuarios', methods=['GET'])
def obtener_usuarios():
    usuarios = [
        {'id': 1, 'nombre': 'Ana'},
        {'id': 2, 'nombre': 'Luis'}
    ]
    return jsonify(usuarios)

En este ejemplo, jsonify convierte la lista usuarios en una respuesta JSON válida, configurando automáticamente el encabezado Content-Type como application/json.

Devolviendo diccionarios y listas

jsonify puede manejar distintos tipos de datos de Python, como diccionarios, listas, y túplas. Esto permite devolver una variedad de estructuras JSON:

@app.route('/api/status', methods=['GET'])
def obtener_status():
    estado = {'status': 'operativo', 'version': '1.0.0'}
    return jsonify(estado)

En este caso, se devuelve un diccionario que representa el estado del sistema.

Especificando códigos de estado HTTP

Por defecto, jsonify devuelve un código de estado 200 OK. Sin embargo, es posible especificar otros códigos de estado añadiéndolos como segundo elemento en la tupla de retorno:

@app.route('/api/usuarios/<int:id>', methods=['GET'])
def obtener_usuario(id):
    usuario = buscar_usuario_por_id(id)
    if usuario:
        return jsonify(usuario), 200
    else:
        return jsonify({'error': 'Usuario no encontrado'}), 404

Aquí, si el usuario no es encontrado, se devuelve una respuesta con código de estado 404 Not Found.

Manejo de errores con jsonify

Es recomendable utilizar jsonify al manejar errores para mantener consistencia en las respuestas JSON:

@app.errorhandler(400)
def peticion_invalida(error):
    return jsonify({'error': 'Petición inválida'}), 400

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

Estos manejadores capturan excepciones y devuelven respuestas JSON con el código de estado apropiado.

Añadiendo mensajes y metadatos

Podemos enriquecer las respuestas incluyendo mensajes y metadatos adicionales:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    productos = obtener_todos_los_productos()
    respuesta = {
        'total': len(productos),
        'productos': productos,
        'mensaje': 'Lista de productos obtenida con éxito'
    }
    return jsonify(respuesta)

En este ejemplo, la respuesta incluye el número total de productos y un mensaje de éxito, además de la lista de productos en sí.

Serialización de objetos complejos

Si necesitamos devolver objetos más complejos, debemos asegurarnos de que sean serializables. Una práctica común es definir un método que convierta el objeto en un diccionario:

class Producto(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100))
    precio = db.Column(db.Float)

    def a_dict(self):
        return {
            'id': self.id,
            'nombre': self.nombre,
            'precio': self.precio
        }

@app.route('/api/productos/<int:id>', methods=['GET'])
def obtener_producto(id):
    producto = Producto.query.get(id)
    if producto:
        return jsonify(producto.a_dict())
    else:
        return jsonify({'error': 'Producto no encontrado'}), 404

El método a_dict facilita la conversión del objeto Producto en un diccionario serializable.

Convertir objetos no serializables

Algunos tipos de datos, como objetos datetime, no son serializables por defecto. Debemos convertirlos a un formato compatible:

from datetime import datetime

@app.route('/api/hora', methods=['GET'])
def obtener_hora():
    ahora = datetime.utcnow()
    return jsonify({'hora_actual': ahora.isoformat() + 'Z'})

Al usar isoformat(), transformamos el objeto datetime en una cadena ISO 8601, que es serializable y ampliamente utilizada.

Uso de make_response para mayor control

En ocasiones, necesitamos mayor control sobre la respuesta HTTP. Podemos utilizar make_response en conjunto con jsonify:

from flask import make_response

@app.route('/api/descarga', methods=['GET'])
def descargar_datos():
    datos = {'clave': 'valor'}
    respuesta = make_response(jsonify(datos), 200)
    respuesta.headers['Content-Disposition'] = 'attachment; filename="datos.json"'
    return respuesta

Esto permite modificar encabezados adicionales, como Content-Disposition, para indicar que la respuesta es un archivo adjunto.

Beneficios de usar jsonify

El uso de jsonify ofrece varias ventajas:

• Configuración automática del encabezado Content-Type.
• Serialización segura de tipos de datos compatibles.
• Prevención de vulnerabilidades, como ataques de inyección.
• Legibilidad y consistencia en el código.

Por estas razones, es una práctica recomendada utilizar jsonify en lugar de devolver manualmente cadenas JSON o utilizar otras funciones de serialización.

Evitar retornar datos sin serializar

Es importante no retornar datos que no sean serializables, ya que causará errores:

@app.route('/api/usuarios_objeto', methods=['GET'])
def obtener_usuario_objeto():
    usuario = Usuario(nombre='Maria', email='maria@example.com')
    return jsonify(usuario)  # Esto generará un error

Para solucionar esto, debemos convertir el objeto Usuario en un diccionario o utilizar un esquema de serialización.

Integración con esquemas de serialización

Para aplicaciones más complejas, podemos utilizar librerías como Marshmallow para definir esquemas de serialización, pero esto se cubrirá en secciones posteriores.

Peticiones GET sin parámetros

Las peticiones GET sin parámetros son esenciales para recuperar listas completas de recursos en una API REST. En Flask, se pueden implementar estas peticiones definiendo rutas que no requieren parámetros adicionales en la URL y que responden al método GET.

Supongamos que tenemos una aplicación que gestiona productos y queremos obtener la lista completa de productos almacenados en la base de datos. Primero, definimos un modelo Producto utilizando SQLAlchemy:

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class Producto(db.Model):
    __tablename__ = 'productos'
    id = db.Column(db.Integer, primary_key=True)
    nombre = db.Column(db.String(100), nullable=False)
    precio = db.Column(db.Float, nullable=False)

    def a_dict(self):
        return {
            'id': self.id,
            'nombre': self.nombre,
            'precio': self.precio
        }

En este modelo, Producto representa una tabla con campos id, nombre y precio. El método a_dict permite convertir una instancia del producto en un diccionario para facilitar su serialización.

A continuación, creamos una ruta en nuestro controlador REST para manejar la petición GET sin parámetros:

from flask import Flask, jsonify
from modelos import db, Producto

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://usuario:contraseña@localhost:3306/tienda'
db.init_app(app)

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    productos = Producto.query.all()
    lista_productos = [producto.a_dict() for producto in productos]
    return jsonify(lista_productos), 200

En este fragmento de código:

• Configuramos la aplicación Flask y establecemos la conexión con la base de datos MySQL.
• Utilizamos @app.route para definir la ruta /api/productos que responde al método GET.
• Obtenemos todos los registros de la tabla productos mediante Producto.query.all().
• Convertimos cada producto en un diccionario utilizando el método a_dict y los agregamos a la lista lista_productos.
• Devolvemos la lista de productos en formato JSON utilizando jsonify, junto con el código de estado 200 OK.

Es importante manejar posibles excepciones que puedan ocurrir durante la consulta a la base de datos para garantizar la robustez de la aplicación:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    try:
        productos = Producto.query.all()
        lista_productos = [producto.a_dict() for producto in productos]
        return jsonify(lista_productos), 200
    except Exception as e:
        return jsonify({'error': 'No se pudieron obtener los productos'}), 500

De esta manera, si ocurre un error, se devuelve una respuesta con el código de estado 500 Internal Server Error y un mensaje de error descriptivo.

Si la cantidad de datos es considerable, es recomendable implementar paginación para optimizar el rendimiento y evitar sobrecargar la respuesta:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    try:
        pagina = request.args.get('pagina', 1, type=int)
        tamanio = request.args.get('tamanio', 10, type=int)
        productos_paginados = Producto.query.paginate(page=pagina, per_page=tamanio)
        lista_productos = [producto.a_dict() for producto in productos_paginados.items]
        respuesta = {
            'total': productos_paginados.total,
            'pagina': productos_paginados.page,
            'paginas': productos_paginados.pages,
            'productos': lista_productos
        }
        return jsonify(respuesta), 200
    except Exception as e:
        return jsonify({'error': 'No se pudieron obtener los productos'}), 500

En este ejemplo:

• Utilizamos request.args.get para obtener los parámetros opcionales pagina y tamanio, con valores predeterminados.
• Aplicamos la paginación con Producto.query.paginate.
• Incluimos metadatos en la respuesta como el número total de productos, la página actual y el número total de páginas, además de la lista de productos.

Si queremos mantener la petición sin parámetros pero con funcionalidad de limitación, podemos definir valores predeterminados:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    try:
        productos = Producto.query.limit(100).all()
        lista_productos = [producto.a_dict() for producto in productos]
        return jsonify(lista_productos), 200
    except Exception as e:
        return jsonify({'error': 'No se pudieron obtener los productos'}), 500

Así, limitamos la cantidad de productos devueltos a 100, evitando posibles problemas de rendimiento sin necesidad de parámetros adicionales.

También es posible agregar cabeceras a la respuesta para proporcionar información extra:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    try:
        productos = Producto.query.all()
        lista_productos = [producto.a_dict() for producto in productos]
        respuesta = jsonify(lista_productos)
        respuesta.headers['X-Total-Count'] = len(lista_productos)
        return respuesta, 200
    except Exception as e:
        return jsonify({'error': 'No se pudieron obtener los productos'}), 500

Al incluir la cabecera X-Total-Count, informamos al cliente del número total de productos devueltos.

Al implementar peticiones GET sin parámetros, es crucial seguir buenas prácticas de seguridad, como proteger los datos sensibles y validar adecuadamente los modelos. Además, es recomendable utilizar Blueprints y una estructura modular para mantener el código limpio y escalable.

Finalmente, recordemos que es necesario ejecutar la aplicación dentro del bloque habitual:

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

Es importante desactivar el modo debug en entornos de producción para garantizar la seguridad de la aplicación.

Con estas consideraciones, podemos manejar eficientemente las peticiones GET sin parámetros en Flask, proporcionando a los clientes acceso a los recursos necesarios de manera segura y optimizada.

Peticiones GET con parámetros

En muchas aplicaciones, las peticiones GET necesitan recibir parámetros para filtrar, ordenar o limitar los datos devueltos. En Flask, los parámetros de consulta se obtienen utilizando el objeto request.args, que es un diccionario inmutable de los parámetros enviados en la URL después del signo de interrogación (?).

Por ejemplo, si tenemos una ruta que devuelve productos y queremos permitir al usuario filtrar por categoría o rango de precio, podemos acceder a esos parámetros de la siguiente manera:

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    categoria = request.args.get('categoria')
    precio_min = request.args.get('precio_min', type=float)
    precio_max = request.args.get('precio_max', type=float)

    consulta = Producto.query

    if categoria:
        consulta = consulta.filter_by(categoria=categoria)
    if precio_min is not None:
        consulta = consulta.filter(Producto.precio >= precio_min)
    if precio_max is not None:
        consulta = consulta.filter(Producto.precio <= precio_max)

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

En este ejemplo:

• Utilizamos request.args.get('categoria') para obtener el parámetro categoria de la URL.
• Especificamos el argumento type=float para convertir los parámetros precio_min y precio_max a números de tipo flotante.
• Construimos dinámicamente la consulta de SQLAlchemy aplicando filtros basados en los parámetros recibidos.
• Devolvemos la lista de productos filtrados en formato JSON.

Uso de parámetros opcionales y valores predeterminados

Es común que los parámetros de consulta sean opcionales. Podemos asignar valores predeterminados si el parámetro no está presente:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    limite = request.args.get('limite', default=10, type=int)
    pagina = request.args.get('pagina', default=1, type=int)

    productos_paginados = Producto.query.paginate(page=pagina, per_page=limite)
    lista_productos = [producto.to_dict() for producto in productos_paginados.items]

    respuesta = {
        'total': productos_paginados.total,
        'pagina': productos_paginados.page,
        'paginas': productos_paginados.pages,
        'productos': lista_productos
    }
    return jsonify(respuesta), 200

En este caso:

• Utilizamos request.args.get con el parámetro default para asignar valores predeterminados a limite y pagina.
• Implementamos paginación utilizando paginate de SQLAlchemy.

Validación de parámetros

Es fundamental validar los parámetros recibidos para evitar errores y garantizar la seguridad. Podemos manejar casos en los que los parámetros no sean del tipo esperado:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    try:
        limite = int(request.args.get('limite', 10))
        pagina = int(request.args.get('pagina', 1))
    except ValueError:
        return jsonify({'error': 'Los parámetros limite y pagina deben ser números enteros'}), 400

    if limite <= 0 or pagina <= 0:
        return jsonify({'error': 'Los parámetros limite y pagina deben ser mayores que cero'}), 400

    productos_paginados = Producto.query.paginate(page=pagina, per_page=limite)
    lista_productos = [producto.to_dict() for producto in productos_paginados.items]
    respuesta = {
        'total': productos_paginados.total,
        'pagina': productos_paginados.page,
        'paginas': productos_paginados.pages,
        'productos': lista_productos
    }
    return jsonify(respuesta), 200

En este fragmento:

• Capturamos excepciones ValueError si los parámetros no pueden convertirse a enteros.
• Verificamos que los valores sean positivos y mayores que cero.
• Devolvemos un código de estado 400 Bad Request con un mensaje de error apropiado en caso de validación fallida.

Parámetros de consulta múltiples

Flask permite acceder a múltiples valores para un mismo parámetro utilizando request.args.getlist. Esto es útil cuando se desea filtrar por varios valores:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    colores = request.args.getlist('color')
    consulta = Producto.query

    if colores:
        consulta = consulta.filter(Producto.color.in_(colores))

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

Si el cliente realiza una petición como /api/productos?color=rojo&color=azul, el parámetro colores será una lista con los valores ['rojo', 'azul'].

Ordenación de resultados

Es posible ordenar los resultados en función de un parámetro de consulta:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    orden = request.args.get('orden', 'nombre')
    direccion = request.args.get('direccion', 'asc')

    if orden not in ['nombre', 'precio']:
        return jsonify({'error': 'Parámetro de ordenación no válido'}), 400

    consulta = Producto.query.order_by(getattr(Producto, orden))
    if direccion == 'desc':
        consulta = consulta.order_by(getattr(Producto, orden).desc())

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

En este ejemplo:

• Validamos que el parámetro orden sea un campo válido para evitar inyecciones de código.
• Utilizamos getattr para obtener el atributo del modelo correspondiente al campo de ordenación.
• Aplicamos la dirección de ordenación según el valor de direccion.

Manejo de parámetros booleanos

Al recibir parámetros booleanos, es necesario interpretarlos correctamente:

def convertir_a_booleano(valor):
    if valor.lower() in ['true', '1', 'si']:
        return True
    elif valor.lower() in ['false', '0', 'no']:
        return False
    else:
        raise ValueError('Valor booleano inválido')

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    disponible = request.args.get('disponible')
    consulta = Producto.query

    if disponible is not=None:
        try:
            es_disponible = convertir_a_booleano(disponible)
            consulta = consulta.filter_by(disponible=es_disponible)
        except ValueError as e:
            return jsonify({'error': str(e)}), 400

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

Aquí:

• Definimos una función convertir_a_booleano para interpretar cadenas como valores booleanos.
• Manejamos posibles errores de validación y devolvemos una respuesta adecuada.

Combinación de parámetros y manejo de errores

Podemos combinar múltiples parámetros y asegurarnos de que la aplicación maneje adecuadamente los errores:

@app.route('/api/productos', methods=['GET'])
def obtener_productos():
    categoria = request.args.get('categoria')
    precio_min = request.args.get('precio_min', type=float)
    precio_max = request.args.get('precio_max', type=float)
    disponible = request.args.get('disponible')
    orden = request.args.get('orden', 'nombre')
    direccion = request.args.get('direccion', 'asc')

    consulta = Producto.query

    if categoria:
        consulta = consulta.filter_by(categoria=categoria)
    if precio_min is not None:
        consulta = consulta.filter(Producto.precio >= precio_min)
    if precio_max is not None:
        consulta = consulta.filter(Producto.precio <= precio_max)
    if disponible is not None:
        try:
            es_disponible = convertir_a_booleano(disponible)
            consulta = consulta.filter_by(disponible=es_disponible)
        except ValueError as e:
            return jsonify({'error': str(e)}), 400
    if orden not in ['nombre', 'precio']:
        return jsonify({'error': 'Parámetro de ordenación no válido'}), 400
    if direccion not in ['asc', 'desc']:
        return jsonify({'error': 'Dirección de ordenación no válida'}), 400

    if direccion == 'asc':
        consulta = consulta.order_by(getattr(Producto, orden).asc())
    else:
        consulta = consulta.order_by(getattr(Producto, orden).desc())

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

Este código:

• Combina filtros por categoría, precio y disponibilidad.
• Incluye ordenación y dirección.
• Valida los parámetros y devuelve mensajes de error claros en caso de valores inválidos.

Seguridad y buenas prácticas

Es crucial seguir buenas prácticas para evitar vulnerabilidades:

• Validar y sanitizar todos los parámetros de entrada.
• Evitar construir consultas SQL directamente con parámetros del usuario para prevenir inyección SQL.
• Manejar excepciones y devolver códigos de estado HTTP apropiados.
• Limitar la exposición de datos sensibles o internos.

Uso de parámetros en rutas

Aunque se cubrió en secciones anteriores el uso de parámetros en las rutas, es posible combinar parámetros en la ruta y parámetros de consulta:

@app.route('/api/categorias/<string:nombre_categoria>/productos', methods=['GET'])
def obtener_productos_por_categoria(nombre_categoria):
    disponible = request.args.get('disponible')
    consulta = Producto.query.filter_by(categoria=nombre_categoria)

    if disponible is not None:
        try:
            es_disponible = convertir_a_booleano(disponible)
            consulta = consulta.filter_by(disponible=es_disponible)
        except ValueError as e:
            return jsonify({'error': str(e)}), 400

    productos = consulta.all()
    lista_productos = [producto.to_dict() for producto in productos]
    return jsonify(lista_productos), 200

De esta manera, los clientes pueden solicitar productos de una categoría específica y aplicar filtros adicionales a través de los parámetros de consulta.