Introducción a async/await

¿Qué es la programación asíncrona?

La programación asíncrona es una forma de escribir código que permite que tu aplicación realice múltiples tareas sin quedarse bloqueada esperando que una operación termine. Imagina que estás cocinando: mientras hierve el agua para la pasta, puedes cortar verduras en lugar de quedarte parado mirando la olla.

En el contexto de las API web, esto significa que tu servidor puede atender múltiples peticiones al mismo tiempo, en lugar de procesar una petición completa antes de pasar a la siguiente.

Diferencia entre código síncrono y asíncrono

Hasta ahora has trabajado con código síncrono en FastAPI. Cuando una función se ejecuta, el programa espera a que termine completamente antes de continuar:

from fastapi import FastAPI
import time

app = FastAPI()

@app.get("/lento")
def operacion_lenta():
    # Simula una operación que tarda 3 segundos
    time.sleep(3)
    return {"mensaje": "Operación completada"}

En este ejemplo, si dos usuarios hacen peticiones al mismo tiempo, el segundo tendrá que esperar a que termine la primera petición antes de ser atendido.

Con programación asíncrona, el comportamiento cambia completamente:

from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get("/rapido")
async def operacion_asincrona():
    # Simula una operación asíncrona de 3 segundos
    await asyncio.sleep(3)
    return {"mensaje": "Operación completada"}

Ahora, mientras una petición está "durmiendo" durante esos 3 segundos, el servidor puede atender otras peticiones sin problema.

Cuándo es útil la programación asíncrona

La asincronía es especialmente valiosa cuando tu API necesita realizar operaciones que implican esperar:

• Consultas a bases de datos: Mientras esperas la respuesta de la base de datos, puedes procesar otras peticiones
• Llamadas a APIs externas: Si tu API consulta otros servicios web, no necesitas bloquear todo el servidor
• Operaciones de archivos: Leer o escribir archivos grandes puede hacerse de forma asíncrona
• Envío de emails: Procesar el envío en segundo plano mientras atiendes más usuarios

import httpx
from fastapi import FastAPI

app = FastAPI()

@app.get("/clima/{ciudad}")
async def obtener_clima(ciudad: str):
    # Llamada asíncrona a una API externa
    async with httpx.AsyncClient() as client:
        respuesta = await client.get(f"https://api.clima.com/{ciudad}")
        datos = respuesta.json()
    
    return {"ciudad": ciudad, "temperatura": datos["temp"]}

Ventajas de la asincronía en FastAPI

FastAPI está diseñado para aprovechar al máximo la programación asíncrona. Las principales ventajas son:

• Mayor rendimiento: Tu API puede manejar más usuarios simultáneamente
• Mejor experiencia de usuario: Los tiempos de respuesta se reducen cuando hay múltiples peticiones
• Uso eficiente de recursos: El servidor no desperdicia tiempo esperando operaciones lentas
• Escalabilidad mejorada: Tu aplicación puede crecer sin necesidad de más hardware inmediatamente

La clave está en entender que la asincronía no hace que las operaciones individuales sean más rápidas, sino que permite que tu servidor sea más eficiente manejando múltiples operaciones al mismo tiempo.

Cuándo NO usar asincronía

No todas las operaciones necesitan ser asíncronas. Si tu función realiza cálculos intensivos que no implican esperar (como procesar números o manipular texto), el código síncrono puede ser más simple y adecuado:

@app.get("/calcular/{numero}")
def calcular_factorial(numero: int):
    # Operación de cálculo puro - no necesita ser asíncrona
    resultado = 1
    for i in range(1, numero + 1):
        resultado *= i
    return {"factorial": resultado}

La regla general es usar asincronía cuando tu código necesita esperar por algo externo: bases de datos, archivos, redes o servicios externos.

Sintaxis básica async/await en FastAPI

Para convertir tus funciones de FastAPI en asíncronas, necesitas usar dos palabras clave de Python: async y await. La sintaxis es sencilla y FastAPI la integra de forma natural.

Declarando funciones asíncronas con async

El primer paso es agregar la palabra async antes de def en tus funciones de ruta:

from fastapi import FastAPI

app = FastAPI()

# Función síncrona (como has usado hasta ahora)
@app.get("/sincrono")
def funcion_sincrona():
    return {"tipo": "síncrono"}

# Función asíncrona (nueva sintaxis)
@app.get("/asincrono")
async def funcion_asincrona():
    return {"tipo": "asíncrono"}

La diferencia principal está en la declaración de la función. Al agregar async, le dices a Python que esta función puede ser pausada y reanudada, permitiendo que otras operaciones se ejecuten mientras tanto.

Usando await para operaciones que requieren espera

La palabra clave await se usa dentro de funciones async para esperar operaciones que pueden tardar tiempo. Solo puedes usar await dentro de funciones declaradas con async:

import asyncio
from fastapi import FastAPI

app = FastAPI()

@app.get("/esperar")
async def operacion_con_espera():
    # await pausa la función hasta que la operación termine
    await asyncio.sleep(2)  # Simula una espera de 2 segundos
    return {"mensaje": "He esperado 2 segundos"}

Ejemplo práctico: consulta a base de datos

Aquí tienes un ejemplo más realista usando una consulta asíncrona a base de datos:

from fastapi import FastAPI
import asyncpg

app = FastAPI()

@app.get("/usuarios/{user_id}")
async def obtener_usuario(user_id: int):
    # Conexión asíncrona a PostgreSQL
    conn = await asyncpg.connect("postgresql://user:pass@localhost/db")
    
    # Consulta asíncrona - el servidor puede atender otras peticiones mientras espera
    usuario = await conn.fetchrow("SELECT * FROM usuarios WHERE id = $1", user_id)
    
    await conn.close()
    
    if usuario:
        return {"id": usuario["id"], "nombre": usuario["nombre"]}
    else:
        return {"error": "Usuario no encontrado"}

Combinando operaciones síncronas y asíncronas

Dentro de una función async, puedes mezclar código normal (síncrono) con operaciones asíncronas:

from fastapi import FastAPI
import httpx
import json

app = FastAPI()

@app.get("/procesar-datos")
async def procesar_datos():
    # Código síncrono normal
    datos_locales = {"timestamp": "2024-01-15", "version": "1.0"}
    
    # Operación asíncrona - llamada a API externa
    async with httpx.AsyncClient() as client:
        respuesta = await client.get("https://api.ejemplo.com/datos")
        datos_externos = respuesta.json()
    
    # Más código síncrono - procesamiento de datos
    resultado = {
        "local": datos_locales,
        "externo": datos_externos,
        "procesado": True
    }
    
    return resultado

Múltiples operaciones await en secuencia

Puedes usar múltiples await en la misma función. Cada uno pausará la ejecución hasta que esa operación específica termine:

@app.get("/multiple-consultas")
async def consultas_multiples():
    async with httpx.AsyncClient() as client:
        # Primera consulta
        clima = await client.get("https://api.clima.com/madrid")
        
        # Segunda consulta (se ejecuta DESPUÉS de la primera)
        noticias = await client.get("https://api.noticias.com/ultimas")
        
        # Tercera consulta
        trafico = await client.get("https://api.trafico.com/madrid")
    
    return {
        "clima": clima.json(),
        "noticias": noticias.json(),
        "trafico": trafico.json()
    }

Validación de datos con Pydantic en funciones async

Los modelos de Pydantic funcionan exactamente igual en funciones asíncronas:

from fastapi import FastAPI
from pydantic import BaseModel
import asyncio

app = FastAPI()

class Usuario(BaseModel):
    nombre: str
    email: str
    edad: int

@app.post("/crear-usuario")
async def crear_usuario(usuario: Usuario):
    # Simula guardar en base de datos de forma asíncrona
    await asyncio.sleep(1)  # Operación de guardado
    
    return {
        "mensaje": "Usuario creado exitosamente",
        "usuario": usuario,
        "id_generado": 12345
    }

Reglas importantes para async/await

Hay algunas reglas básicas que debes recordar:

• Solo usa await dentro de funciones declaradas con async
• No mezcles time.sleep() con funciones async - usa asyncio.sleep() en su lugar
• FastAPI detecta automáticamente si tu función es async o no, y la maneja apropiadamente
• Las funciones async siempre devuelven un objeto especial llamado "corrutina", pero FastAPI se encarga de esto por ti

# ❌ Incorrecto - time.sleep bloquea todo el servidor
@app.get("/mal-ejemplo")
async def ejemplo_incorrecto():
    import time
    time.sleep(2)  # Esto bloquea el servidor completo
    return {"mensaje": "Mal ejemplo"}

# ✅ Correcto - asyncio.sleep permite que otras operaciones continúen
@app.get("/buen-ejemplo")
async def ejemplo_correcto():
    await asyncio.sleep(2)  # Esto NO bloquea el servidor
    return {"mensaje": "Buen ejemplo"}

Con esta sintaxis básica de async y await, ya puedes empezar a crear APIs más eficientes que pueden manejar múltiples peticiones simultáneamente sin bloquearse.