Async y await en Flask

Soporte de async en Flask 3.1

Flask 3.1 ofrece soporte oficial para funciones asíncronas (async def) en vistas, manejadores de errores, señales y hooks del ciclo de vida como before_request y after_request. Esto permite escribir código no bloqueante para operaciones de entrada/salida (I/O) como llamadas a APIs externas, consultas a bases de datos asíncronas o lectura de archivos.

Para activar el soporte completo de async, instala Flask con el extra async:

pip install "Flask[async]"

Esto instala también asgiref, que permite a Flask correr funciones async dentro de su entorno WSGI síncrono.

Vistas asíncronas básicas

Las vistas asíncronas se definen exactamente como las síncronas, pero usando async def:

from flask import Flask, jsonify
import asyncio

app = Flask(__name__)

# Vista síncrona (tradicional)
@app.route('/productos/sync')
def listar_productos_sync():
    productos = obtener_productos_de_db()  # Bloquea el hilo
    return jsonify(productos)

# Vista asíncrona (Flask 3.1)
@app.route('/productos/async')
async def listar_productos_async():
    productos = await obtener_productos_db_async()  # No bloquea
    return jsonify(productos)

El caso más común para usar async es cuando la vista necesita esperar múltiples operaciones de I/O de forma concurrente:

import asyncio
import httpx
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/dashboard')
async def dashboard():
    """
    Obtiene datos de múltiples fuentes concurrentemente.
    Sin async, estas llamadas serían secuenciales y lentas.
    """
    async with httpx.AsyncClient() as cliente:
        # Lanzar todas las peticiones concurrentemente
        usuario_task = cliente.get('https://api.ejemplo.com/usuario/1')
        pedidos_task = cliente.get('https://api.ejemplo.com/pedidos?usuario=1')
        notificaciones_task = cliente.get('https://api.ejemplo.com/notificaciones/1')

        # Esperar todas a la vez (en paralelo)
        usuario_resp, pedidos_resp, notif_resp = await asyncio.gather(
            usuario_task,
            pedidos_task,
            notificaciones_task
        )

    return jsonify({
        'usuario': usuario_resp.json(),
        'pedidos': pedidos_resp.json(),
        'notificaciones': notif_resp.json()
    })

En una vista síncrona, las tres llamadas se harían secuencialmente (p. ej. 300ms + 200ms + 150ms = 650ms). Con asyncio.gather, se ejecutan en paralelo y el tiempo total es el de la más lenta (~300ms).

Cliente HTTP asíncrono con httpx

httpx es la biblioteca HTTP recomendada para código asíncrono en Flask. Es compatible tanto con uso síncrono como asíncrono:

pip install httpx

import httpx
from flask import Flask, jsonify, request

app = Flask(__name__)

# Configurar cliente httpx reutilizable (mejor rendimiento)
cliente_http = httpx.AsyncClient(
    timeout=10.0,
    headers={'User-Agent': 'MiApp-Flask/1.0'}
)

@app.route('/buscar-producto')
async def buscar_producto():
    """Busca un producto en una API externa."""
    termino = request.args.get('q', '')

    if not termino:
        return jsonify({'error': 'Parámetro q requerido'}), 400

    try:
        respuesta = await cliente_http.get(
            'https://api.ejemplo.com/productos/buscar',
            params={'q': termino, 'limit': 10}
        )
        respuesta.raise_for_status()
        return jsonify(respuesta.json())

    except httpx.HTTPStatusError as e:
        return jsonify({
            'error': f'Error en API externa: {e.response.status_code}'
        }), 502
    except httpx.RequestError as e:
        return jsonify({'error': 'No se pudo conectar al servicio'}), 503

@app.route('/convertir-moneda')
async def convertir_moneda():
    """Convierte una cantidad entre divisas usando una API externa."""
    cantidad = float(request.args.get('cantidad', 0))
    origen = request.args.get('de', 'EUR')
    destino = request.args.get('a', 'USD')

    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f'https://api.exchangerate.host/convert',
            params={'from': origen, 'to': destino, 'amount': cantidad}
        )
        datos = resp.json()

    return jsonify({
        'cantidad_original': cantidad,
        'moneda_origen': origen,
        'cantidad_convertida': datos.get('result'),
        'moneda_destino': destino
    })

Hooks asíncronos del ciclo de vida

Los hooks before_request, after_request y teardown_request también pueden ser asíncronos en Flask 3.1:

from flask import Flask, g, request
import time

app = Flask(__name__)

@app.before_request
async def verificar_autorizacion():
    """Hook asíncrono que verifica el token en una base de datos."""
    token = request.headers.get('Authorization', '').replace('Bearer ', '')

    if token and request.path.startswith('/api/'):
        # Verificación asíncrona contra base de datos o Redis
        usuario = await verificar_token_async(token)
        g.usuario = usuario

@app.after_request
async def registrar_telemetria(response):
    """Registra métricas de forma asíncrona."""
    # Enviar a servicio de telemetría sin bloquear
    await enviar_metrica_async(
        endpoint=request.path,
        metodo=request.method,
        codigo=response.status_code
    )
    return response

Limitaciones importantes del async en Flask

Flask es un framework WSGI síncrono con soporte añadido para async. Esto tiene implicaciones importantes:

# CORRECTO: La espera asíncrona es útil para I/O externo
@app.route('/datos-externos')
async def datos_externos():
    async with httpx.AsyncClient() as c:
        resp = await c.get('https://api.externa.com/datos')
    return resp.text

# LIMITACIÓN: Flask sigue siendo WSGI, un hilo por petición
# El async de Flask no escala como FastAPI o Starlette
# Para alto rendimiento con async, considera usar:
# - ASGI con Hypercorn o Uvicorn
# - FastAPI para APIs puras

# INCORRECTO: Mezclar código bloqueante dentro de async
@app.route('/mal-ejemplo')
async def mal_ejemplo():
    import time
    time.sleep(5)  # ¡BLOQUEA el hilo aunque sea async!
    return 'malo'

# CORRECTO: Usar asyncio.sleep en lugar de time.sleep
@app.route('/buen-ejemplo')
async def buen_ejemplo():
    await asyncio.sleep(5)  # No bloquea
    return 'bien'

Comparativa: cuándo usar async en Flask

# BUENA razón para usar async:
# Múltiples llamadas a APIs externas concurrentes
@app.route('/resumen-usuario/<int:uid>')
async def resumen_usuario(uid):
    async with httpx.AsyncClient() as c:
        perfil, pedidos, recomendaciones = await asyncio.gather(
            c.get(f'https://api.usuarios.com/{uid}'),
            c.get(f'https://api.pedidos.com/usuario/{uid}'),
            c.get(f'https://api.recomendaciones.com/usuario/{uid}')
        )
    return jsonify({
        'perfil': perfil.json(),
        'pedidos': pedidos.json(),
        'recomendaciones': recomendaciones.json()
    })

# MALA razón para usar async (no hay I/O asíncrono):
@app.route('/suma')
async def suma():
    resultado = 2 + 2  # Operación CPU, no se beneficia de async
    return jsonify({'resultado': resultado})

Ejecutar Flask con servidor ASGI

Para aprovechar al máximo las vistas asíncronas, Flask puede ejecutarse sobre un servidor ASGI como Hypercorn:

pip install hypercorn

# app.py
from flask import Flask

app = Flask(__name__)

@app.route('/')
async def inicio():
    return 'Flask con ASGI'

# Ejecutar con Hypercorn (servidor ASGI)
hypercorn app:app --bind 0.0.0.0:5000 --workers 4

Con un servidor ASGI, Flask puede manejar múltiples peticiones concurrentes en el mismo proceso, aprovechando mejor las vistas asíncronas cuando hay operaciones de I/O que esperan.

El soporte de async/await en Flask 3.1 es ideal para modernizar aplicaciones existentes añadiendo concurrencia en puntos estratégicos, sin necesidad de migrar a un framework diferente.