Las operaciones asíncronas necesitan límites. Una llamada de red puede no responder, una cola puede permanecer vacía y una dependencia lenta puede consumir todo el presupuesto de una solicitud. asyncio.timeout() crea un ámbito con deadline: si el bloque no termina a tiempo, la tarea actual se cancela y el contexto convierte esa cancelación en TimeoutError.
Esta guía cubre deadlines relativos y absolutos, reagendamiento, comprobación de expiración, integración con TaskGroup, comparación con asyncio.wait_for(), cleanup, shield, reintentos y errores comunes de cancelación.
Primer timeout
import asyncio
async def operacion_lenta() -> str:
await asyncio.sleep(5)
return "ok"
async def main():
try:
async with asyncio.timeout(1.0):
resultado = await operacion_lenta()
print(resultado)
except TimeoutError:
print("plazo excedido")
asyncio.run(main())Después de un segundo, el contexto cancela la tarea actual. Al salir del bloque, la cancelación interna se convierte en TimeoutError, que debe capturarse fuera del contexto.
Por qué capturar fuera
La transformación de CancelledError a TimeoutError ocurre cuando el context manager termina.
try:
async with asyncio.timeout(1.0):
await operacion()
except TimeoutError:
...El patrón también deja claro qué ámbito posee el deadline.
Un timeout no interrumpe código bloqueante
La cancelación de asyncio es cooperativa. El timeout puede actuar únicamente cuando la coroutine entrega el control al event loop. Código CPU-bound o una llamada bloqueante directa impide reaccionar.
async def incorrecto():
time.sleep(10) # bloquea el event loopUsa asyncio.to_thread(), un executor o un proceso.
async with asyncio.timeout(2.0):
resultado = await asyncio.to_thread(funcion_bloqueante)Cancelar el await deja de esperar la thread, pero no necesariamente detiene la función subyacente. Configura límites nativos en la biblioteca bloqueante cuando sea posible.
Timeout relativo y deadline absoluto
asyncio.timeout(segundos) define un límite relativo. asyncio.timeout_at(cuando) recibe un instante absoluto del reloj monotónico del loop.
loop = asyncio.get_running_loop()
deadline = loop.time() + 3.0
async with asyncio.timeout_at(deadline):
await etapa_a()
await etapa_b()Los deadlines absolutos son útiles cuando varias capas deben compartir un único presupuesto sin reiniciar el cronómetro.
Propagar un presupuesto de tiempo
Si una solicitud dispone de cinco segundos, cada función interna no debería recibir cinco segundos nuevos.
async def consultar_a(deadline: float):
async with asyncio.timeout_at(deadline):
return await llamada_a()
async def flujo():
loop = asyncio.get_running_loop()
deadline = loop.time() + 5.0
a = await consultar_a(deadline)
b = await consultar_b(deadline)
return a, bEl tiempo gastado en la primera etapa reduce automáticamente lo disponible para la segunda.
Reagendar el timeout
El objeto del contexto ofrece reschedule(). Es útil cuando el deadline se conoce después de leer metadatos o negociar con otro sistema.
async def procesar():
loop = asyncio.get_running_loop()
async with asyncio.timeout(None) as control:
limite = await obtener_limite()
control.reschedule(loop.time() + limite)
return await ejecutar_trabajo()None crea un contexto inicialmente sin plazo. El nuevo instante usa el reloj monotónico del loop.
Comprobar expiración
Después del contexto, expired() indica si se alcanzó el deadline.
control = None
try:
async with asyncio.timeout(1.0) as control:
await trabajo()
except TimeoutError:
pass
if control is not None and control.expired():
print("el plazo expiró")Capturar TimeoutError suele ser suficiente. expired() ayuda en métricas y diagnóstico.
Timeouts anidados
Los contextos pueden anidarse. El deadline más corto se activa primero.
async with asyncio.timeout(10.0):
await etapa_rapida()
try:
async with asyncio.timeout(1.0):
await llamada_opcional()
except TimeoutError:
usar_fallback()
await etapa_final()El timeout interno puede manejarse localmente y el externo sigue protegiendo el presupuesto total.
Combinar con TaskGroup
Un timeout alrededor de TaskGroup protege toda la unidad concurrente.
async def panel():
try:
async with asyncio.timeout(3.0):
async with asyncio.TaskGroup() as grupo:
usuario = grupo.create_task(obtener_usuario())
pedidos = grupo.create_task(obtener_pedidos())
except TimeoutError:
return {"error": "servicios lentos"}
return {
"usuario": usuario.result(),
"pedidos": pedidos.result(),
}Al expirar, se cancela la tarea contenedora; TaskGroup cancela las hijas y espera su cleanup.
Límites individuales dentro del grupo
Coloca un timeout dentro de cada hija cuando las operaciones tengan límites distintos.
async def consultar_con_limite(nombre: str, segundos: float):
async with asyncio.timeout(segundos):
return await consultar(nombre)Si TimeoutError escapa, TaskGroup trata la tarea como fallida y cancela las hermanas. Si el timeout es esperado, captúralo dentro y devuelve un estado explícito.
Comparación con asyncio.wait_for
asyncio.wait_for(awaitable, timeout) envuelve un awaitable. asyncio.timeout() envuelve un bloque completo con varios awaits.
resultado = await asyncio.wait_for(operacion(), timeout=2.0)
async with asyncio.timeout(2.0):
cabecera = await leer_cabecera()
cuerpo = await leer_cuerpo(cabecera)
await validar(cuerpo)El context manager suele ser más claro para flujos compuestos. wait_for() sigue siendo cómodo para una operación.
Cleanup y finally
Las coroutines canceladas deben liberar recursos.
async def usar_conexion():
conexion = await abrir_conexion()
try:
return await conexion.recibir()
finally:
await conexion.cerrar()El timeout espera mientras la cancelación se propaga y el cleanup termina. Un cierre lento puede hacer que la duración real supere el plazo nominal. Mantén los cierres previsibles.
No ocultar CancelledError
Dentro de la operación, el timeout funciona mediante cancelación. Capturar CancelledError y retornar normalmente puede impedir que el contexto detecte la expiración.
async def malo():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return "ignorado"Realiza cleanup y vuelve a lanzar.
except asyncio.CancelledError:
await limpiar()
raiseShield para trabajo crítico
asyncio.shield() puede proteger una tarea interna de la cancelación, pero la propiedad debe quedar clara.
tarea = asyncio.create_task(escribir_auditoria())
try:
async with asyncio.timeout(1.0):
await asyncio.shield(tarea)
except TimeoutError:
...
await tareaEl timeout deja de esperar, pero la tarea continúa. Conserva una referencia fuerte y define quién la esperará. Shield puede crear trabajo suelto si se usa sin supervisión.
Timeout no es retry
Un timeout termina un intento lento. La política de reintentos debe definir cantidad, backoff, jitter, idempotencia y presupuesto total.
async def intentar(deadline: float):
for indice in range(3):
try:
async with asyncio.timeout_at(deadline):
return await llamada()
except TimeoutError:
if indice == 2:
raise
await asyncio.sleep(0.1 * (2 ** indice))El deadline absoluto impide que los reintentos superen el presupuesto de la solicitud.
Pruebas de timeout
Evita tests con pausas largas. Usa límites breves pero razonables, asyncio.Event y fakes controlados.
async def nunca(evento: asyncio.Event):
await evento.wait()
async def prueba_timeout():
evento = asyncio.Event()
try:
async with asyncio.timeout(0.05):
await nunca(evento)
except TimeoutError:
passNo verifiques milisegundos exactos en CI. Prueba el comportamiento: excepción, cleanup y cancelación de tareas relacionadas.
Métricas y observabilidad
Diferencia timeout de conexión, lectura, cola, operación y presupuesto total. Registra duración, presupuesto restante, dependencia e identificador de solicitud.
Una tasa elevada puede indicar saturación, una dependencia lenta, un límite demasiado agresivo o un event loop bloqueado.
Errores comunes
- Capturar TimeoutError dentro del contexto: la conversión ocurre al salir.
- Usar time.sleep en código async: bloquea el loop.
- Ocultar CancelledError: rompe la semántica del timeout.
- Reiniciar el presupuesto en cada capa: usa deadlines absolutos.
- Esperar que el timeout mate una thread: la cancelación async no detiene trabajo nativo arbitrario.
- Usar shield sin ownership: el trabajo puede continuar sin supervisión.
Ejemplo completo: presupuesto compartido
import asyncio
async def consultar_servicio(nombre: str, deadline: float) -> dict:
async with asyncio.timeout_at(deadline):
await asyncio.sleep(0.2)
return {"servicio": nombre, "ok": True}
async def agregar() -> list[dict]:
loop = asyncio.get_running_loop()
deadline = loop.time() + 2.0
tareas: list[asyncio.Task[dict]] = []
async with asyncio.timeout_at(deadline):
async with asyncio.TaskGroup() as grupo:
for nombre in ["perfil", "pedidos", "saldo"]:
tareas.append(
grupo.create_task(
consultar_servicio(nombre, deadline),
name=f"consulta-{nombre}",
)
)
return [tarea.result() for tarea in tareas]Todas las operaciones comparten el mismo deadline y pertenecen al mismo grupo estructurado.
Conclusión
asyncio.timeout() convierte un deadline en un ámbito claro. Usa cancelación cooperativa, transforma la expiración en TimeoutError y puede proteger varias operaciones relacionadas.
La documentación oficial de timeouts de asyncio detalla timeout(), timeout_at(), reagendamiento y expiración. Usa deadlines absolutos para presupuestos compartidos, preserva cleanup y trata los timeouts como parte de una política de resiliencia.
