Middleware y hooks del ciclo de vida en Flask

El ciclo de vida de una petición en Flask

Entender el ciclo de vida de una petición en Flask es fundamental para implementar comportamientos transversales como autenticación, logging, gestión de conexiones y métricas. Cada petición HTTP pasa por varias fases bien definidas.

El orden de ejecución en Flask es:

1. Se recibe la petición HTTP del servidor WSGI
2. Se activa el contexto de aplicación (app_context)
3. Se activa el contexto de petición (request_context)
4. Se ejecutan las funciones decoradas con @app.before_request
5. Se ejecuta la función de vista correspondiente a la ruta
6. Se ejecutan las funciones decoradas con @app.after_request
7. Se envía la respuesta al cliente
8. Se ejecutan las funciones decoradas con @app.teardown_request
9. Se desactiva el contexto de petición

from flask import Flask, request, g
import time
import logging

app = Flask(__name__)
logger = logging.getLogger(__name__)

@app.before_request
def antes_de_la_peticion():
    """Se ejecuta ANTES de cada petición."""
    g.tiempo_inicio = time.time()
    logger.info(f'Petición: {request.method} {request.path} desde {request.remote_addr}')

@app.after_request
def despues_de_la_peticion(response):
    """Se ejecuta DESPUÉS de cada petición. DEBE devolver la respuesta."""
    duracion = time.time() - g.get('tiempo_inicio', time.time())
    logger.info(f'Respuesta: {response.status_code} en {duracion:.3f}s')
    # Añadir cabecera personalizada con el tiempo de procesamiento
    response.headers['X-Tiempo-Procesamiento'] = f'{duracion:.3f}s'
    return response

@app.teardown_request
def limpiar_peticion(excepcion=None):
    """Se ejecuta siempre al final, incluso si hubo excepción."""
    if excepcion:
        logger.error(f'Excepción durante la petición: {excepcion}')
    # Liberar recursos aquí (conexiones de base de datos, etc.)

El objeto g: datos globales de la petición

Flask proporciona el objeto g (global del contexto de petición) para compartir datos entre las diferentes fases del ciclo de vida de una petición:

from flask import Flask, g, request, jsonify
from functools import wraps

app = Flask(__name__)

def obtener_usuario_actual():
    """Carga el usuario actual desde el token de la petición."""
    if not hasattr(g, 'usuario'):
        token = request.headers.get('Authorization', '').replace('Bearer ', '')
        if token:
            from app.models import Usuario
            from app.servicios.jwt import verificar_token
            payload = verificar_token(token)
            if payload:
                g.usuario = db.session.get(Usuario, payload.get('user_id'))
            else:
                g.usuario = None
        else:
            g.usuario = None
    return g.usuario

@app.before_request
def cargar_usuario():
    """Carga el usuario en g para todas las peticiones."""
    g.usuario = obtener_usuario_actual()

def login_requerido(f):
    """Decorador que verifica autenticación usando g.usuario."""
    @wraps(f)
    def funcion_decorada(*args, **kwargs):
        if g.usuario is None:
            return jsonify({'error': 'Autenticación requerida'}), 401
        return f(*args, **kwargs)
    return funcion_decorada

@app.route('/api/perfil')
@login_requerido
def perfil():
    return jsonify({
        'id': g.usuario.id,
        'nombre': g.usuario.nombre,
        'email': g.usuario.email
    })

before_request y after_request en Blueprints

Los hooks también funcionan a nivel de Blueprint, afectando solo a las rutas de ese módulo:

from flask import Blueprint, g, request, jsonify
import time

api_bp = Blueprint('api', __name__, url_prefix='/api')

@api_bp.before_request
def verificar_api_key():
    """Verifica la API key solo para rutas del Blueprint api."""
    api_key = request.headers.get('X-API-Key')
    if not api_key:
        return jsonify({'error': 'API key requerida'}), 401

    from app.models import ApiKey
    from app.extensions import db
    clave = db.session.execute(
        db.select(ApiKey).filter_by(clave=api_key, activa=True)
    ).scalar_one_or_none()
    if not clave:
        return jsonify({'error': 'API key inválida o desactivada'}), 403

    g.api_key = clave
    g.tiempo_inicio = time.time()

@api_bp.after_request
def registrar_uso_api(response):
    """Registra el uso de la API para estadísticas."""
    if hasattr(g, 'api_key'):
        from app.models import LogApiKey
        from app.extensions import db

        log = LogApiKey(
            api_key_id=g.api_key.id,
            endpoint=request.path,
            metodo=request.method,
            codigo_respuesta=response.status_code,
            duracion_ms=int((time.time() - g.tiempo_inicio) * 1000)
        )
        db.session.add(log)
        db.session.commit()

    return response

@api_bp.route('/productos')
def listar_productos():
    return jsonify({'productos': []})

Middleware WSGI personalizado

El middleware WSGI envuelve la aplicación Flask a nivel del protocolo WSGI, permitiendo interceptar peticiones antes de que Flask las procese:

# middleware/logging_middleware.py
import time
import logging

logger = logging.getLogger(__name__)

class LoggingMiddleware:
    """
    Middleware WSGI que registra todas las peticiones y respuestas.
    """
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        # Información de la petición
        metodo = environ.get('REQUEST_METHOD', 'DESCONOCIDO')
        ruta = environ.get('PATH_INFO', '/')
        inicio = time.time()

        logger.info(f'[MIDDLEWARE] --> {metodo} {ruta}')

        # Capturar el código de estado de la respuesta
        estado_respuesta = {}

        def start_response_modificado(status, headers, exc_info=None):
            estado_respuesta['status'] = status
            return start_response(status, headers, exc_info)

        # Pasar al siguiente middleware o a Flask
        resultado = self.app(environ, start_response_modificado)

        duracion = time.time() - inicio
        logger.info(
            f'[MIDDLEWARE] <-- {metodo} {ruta} '
            f'{estado_respuesta.get("status", "?")} '
            f'({duracion:.3f}s)'
        )

        return resultado

class CabecerasSeguridad:
    """Middleware que añade cabeceras de seguridad a todas las respuestas."""

    CABECERAS = {
        'X-Content-Type-Options': 'nosniff',
        'X-Frame-Options': 'SAMEORIGIN',
        'X-XSS-Protection': '1; mode=block',
        'Referrer-Policy': 'strict-origin-when-cross-origin',
        'Permissions-Policy': 'geolocation=(), microphone=()',
    }

    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        def start_response_con_seguridad(status, headers, exc_info=None):
            headers_lista = list(headers)
            for nombre, valor in self.CABECERAS.items():
                headers_lista.append((nombre, valor))
            return start_response(status, headers_lista, exc_info)

        return self.app(environ, start_response_con_seguridad)

Aplicar el middleware a la aplicación Flask:

# app.py
from flask import Flask
from middleware.logging_middleware import LoggingMiddleware, CabecerasSeguridad

app = Flask(__name__)

# Envolver la aplicación con middleware WSGI
app.wsgi_app = LoggingMiddleware(app.wsgi_app)
app.wsgi_app = CabecerasSeguridad(app.wsgi_app)

Señales de Flask con Blinker

Las señales de Flask permiten notificar y reaccionar a eventos internos sin acoplar directamente los componentes. Flask usa la biblioteca Blinker para implementar señales:

pip install blinker

Flask incluye señales predefinidas que puedes suscribir:

from flask import Flask
from flask.signals import (
    request_started,
    request_finished,
    request_tearing_down,
    got_request_exception,
    appcontext_pushed,
    appcontext_popped
)

app = Flask(__name__)

@request_started.connect_via(app)
def al_iniciar_peticion(sender, **extra):
    """Se ejecuta cuando inicia el procesamiento de una petición."""
    print(f'Señal: petición iniciada en {sender.name}')

@request_finished.connect_via(app)
def al_finalizar_peticion(sender, response, **extra):
    """Se ejecuta cuando la petición termina con éxito."""
    print(f'Señal: petición finalizada con código {response.status_code}')

@got_request_exception.connect_via(app)
def al_recibir_excepcion(sender, exception, **extra):
    """Se ejecuta cuando ocurre una excepción no manejada."""
    import traceback
    print(f'Señal: excepción en petición: {exception}')
    traceback.print_exc()

Señales personalizadas

Puedes crear tus propias señales para desacoplar componentes de tu aplicación:

# app/signals.py
from blinker import Namespace

senales_app = Namespace()

# Definir señales personalizadas
usuario_registrado = senales_app.signal('usuario-registrado')
pago_procesado = senales_app.signal('pago-procesado')
archivo_subido = senales_app.signal('archivo-subido')

# Emitir señales desde la lógica de negocio
from app.signals import usuario_registrado, pago_procesado

@app.route('/api/usuarios/registrar', methods=['POST'])
def registrar_usuario():
    datos = request.get_json()
    usuario = crear_usuario(datos)

    # Emitir señal después de crear el usuario
    usuario_registrado.send(app, usuario=usuario)

    return jsonify(usuario.to_dict()), 201

# Suscribirse a la señal para enviar email de bienvenida
@usuario_registrado.connect_via(app)
def enviar_bienvenida(sender, usuario, **extra):
    from app.servicios.email import enviar_email
    enviar_email(
        destinatario=usuario.email,
        asunto='Bienvenido a nuestra plataforma',
        cuerpo=f'Hola {usuario.nombre}, tu cuenta ha sido creada.'
    )

# Suscribirse para registrar en el log de auditoría
@usuario_registrado.connect_via(app)
def registrar_auditoria(sender, usuario, **extra):
    from app.models import LogAuditoria
    from app.extensions import db
    log = LogAuditoria(
        accion='USUARIO_REGISTRADO',
        entidad_id=usuario.id,
        descripcion=f'Nuevo usuario: {usuario.email}'
    )
    db.session.add(log)
    db.session.commit()

Contexto de aplicación y teardown

Flask separa el contexto de aplicación del contexto de petición. El contexto de aplicación se usa para tareas fuera del ciclo de petición/respuesta:

from flask import Flask
from app.extensions import db

app = Flask(__name__)

@app.teardown_appcontext
def cerrar_conexion_db(excepcion=None):
    """Libera la conexión de base de datos al cerrar el contexto de app."""
    db.session.remove()

# Uso del contexto de aplicación en scripts
with app.app_context():
    from app.models import Usuario
    from app.extensions import db
    usuarios = db.session.execute(db.select(Usuario)).scalars().all()
    for u in usuarios:
        print(f'{u.nombre}: {u.email}')

Los hooks y el middleware son herramientas de programación orientada a aspectos (AOP) que permiten implementar comportamiento transversal (logging, seguridad, métricas) de forma limpia y sin contaminar la lógica de negocio principal.