Conceptos básicos de Asyncio

9 ejemplos para empezar con Asyncio: 6 básicos y 3 intermedios.

Prerrequisitos

• Python 3.14.0 con la biblioteca estándar asyncio.
• Utiliza la sintaxis async/await, no los generadores heredados @asyncio.coroutine.

Ejemplos básicos

1. Definición de corrutina

Las corrutinas son funciones asíncronas pausadas en await.

import asyncio
 
async def hello() -> str:
    await asyncio.sleep(0.1)
    return "hello"
 
async def main() -> None:
    print(await hello())
 
asyncio.run(main())

• async def define una función corrutina.
• Llamarla devuelve un objeto corrutina; debe ser esperado (awaited) o programado.
• asyncio.run gestiona el ciclo de vida del bucle para scripts.

Relacionado: Tareas y Gather - corrutinas concurrentes

2. await pausa cooperativamente

Otras tareas se ejecutan mientras una espera E/S.

import asyncio
 
async def tick(name: str) -> None:
    print(name, "start")
    await asyncio.sleep(0.1)
    print(name, "end")
 
async def main() -> None:
    await tick("a")
    await tick("b")
 
asyncio.run(main())

• await secuencial ejecuta una tras otra.
• sleep cede el control al bucle de eventos.
• No hay paralelismo en CPU, solo programación cooperativa.

Relacionado: Depuración de código asíncrono - errores de bloqueo

3. gather para concurrencia

Ejecuta múltiples corrutinas de forma concurrente.

import asyncio
 
async def work(n: int) -> int:
    await asyncio.sleep(0.1)
    return n
 
async def main() -> None:
    results = await asyncio.gather(work(1), work(2), work(3))
    print(results)
 
asyncio.run(main())

• gather programa todos los awaitables en el mismo bucle.
• Devuelve una lista en el orden de envío cuando todos tienen éxito.
• Un fallo cancela los demás, a menos que se use return_exceptions=True.

Relacionado: Tareas y Gather - objetos Task

4. Programación create_task de "disparar y olvidar"

Programa trabajo en segundo plano sin esperarlo inmediatamente.

import asyncio
 
async def background() -> None:
    await asyncio.sleep(0.1)
    print("done")
 
async def main() -> None:
    task = asyncio.create_task(background())
    print("scheduled")
    await task
 
asyncio.run(main())

• create_task envuelve la corrutina en una Tarea (Task).
• Las Tareas se ejecutan concurrentemente hasta que se esperan o el bucle se detiene.
• Almacena referencias a las tareas para evitar la recolección de basura en pleno vuelo (caso límite raro).

Relacionado: Concurrencia Estructurada - TaskGroup

5. asyncio.sleep vs time.sleep

Nunca bloquees el bucle con sleep síncrono.

import asyncio
 
async def good() -> None:
    await asyncio.sleep(0.05)
 
# time.sleep(0.05) en una corrutina bloquea TODAS las tareas; evítalo

• asyncio.sleep es esperable (awaitable) y no bloquea otras tareas.
• time.sleep congela todo el hilo del bucle de eventos.
• La misma regla se aplica a clientes HTTP síncronos, drivers de bases de datos síncronos en corrutinas.

Relacionado: Puente Síncrono/Asíncrono - descarga de trabajo bloqueante

6. Tiempo de espera con wait_for

Limita el tiempo de espera en corrutinas lentas.

import asyncio
 
async def slow() -> str:
    await asyncio.sleep(1)
    return "ok"
 
async def main() -> None:
    try:
        await asyncio.wait_for(slow(), timeout=0.1)
    except TimeoutError:
        print("timed out")
 
asyncio.run(main())

• Lanza TimeoutError al expirar.
• Cancela la tarea subyacente al agotar el tiempo de espera.
• Prefiere el gestor de contexto asyncio.timeout (3.11+) en código nuevo.

Relacionado: Concurrencia Estructurada - patrones de tiempo de espera

Ejemplos intermedios

7. Patrón de recurso async with

Los gestores de contexto asíncronos liberan recursos sin bloquear a los compañeros.

import asyncio
 
class AsyncResource:
    async def __aenter__(self):
        await asyncio.sleep(0)
        return self
 
    async def __aexit__(self, *args):
        await asyncio.sleep(0)
 
async def main() -> None:
    async with AsyncResource() as res:
        print("using", res)
 
asyncio.run(main())

• __aenter__/__aexit__ reflejan el with síncrono.
• Úsalo para conexiones, bloqueos (locks), sesiones.
• Consulta la página dedicada para iteradores de streaming.

Relacionado: Gestores de Contexto e Iteradores Asíncronos

8. Olor a trabajo de CPU ligado al bucle

Los bucles de CPU privan de recursos a otras corrutinas.

import asyncio
 
async def bad_cpu() -> int:
    total = 0
    for i in range(500_000):
        total += i
    return total
 
# Solución: await asyncio.to_thread(cpu_intensive_fn)

• Las secciones síncronas largas bloquean el "latido" y las comprobaciones de estado.
• Descarga el trabajo a un pool de hilos o procesos.
• Perfila antes de optimizar; los bucles pequeños están bien.

Relacionado: Puente Síncrono/Asíncrono - to_thread

9. Entrada de ciclo de vida de FastAPI

Los servidores mantienen un único bucle de larga duración; no se usa asyncio.run repetidamente por solicitud.

# Patrón conceptual de FastAPI 0.115+
# @asynccontextmanager async def lifespan(app):
#     yield
# app = FastAPI(lifespan=lifespan)

• El framework se encarga de la vida útil del bucle.
• Abre un httpx.AsyncClient compartido al inicio del ciclo de vida.
• Cierra los recursos al apagar.

Relacionado: HTTP Asíncrono con httpx/aiohttp - cliente compartido

Versiones de la pila: Esta página se escribió para Python 3.14.0 (estable 3.14, mantenimiento 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ y uv 0.6+.