asyncio.Runner ofrece una forma estructurada de ejecutar varias corrutinas de nivel superior con el mismo event loop. Es útil en programas de terminal, herramientas administrativas, pruebas, scripts de integración y aplicaciones que necesitan invocar código asíncrono más de una vez sin reconstruir toda la infraestructura en cada llamada. Aunque asyncio.run() sigue siendo la opción más sencilla para un único punto de entrada asíncrono, Runner proporciona mayor control cuando el programa contiene fases diferentes.
Esta guía explica cómo crear un Runner, reutilizar su loop, conservar variables de contexto, tratar señales, activar el modo debug, cerrar recursos y evitar errores frecuentes. El objetivo no es mostrar solamente la sintaxis, sino también un diseño confiable para aplicaciones reales.
Por qué existe asyncio.Runner
asyncio.run() crea un nuevo event loop, ejecuta una corrutina, finaliza generadores asíncronos y cierra el loop. Ese comportamiento es ideal para un programa con una única función principal. Sin embargo, una herramienta puede necesitar cargar configuración, sincronizar datos y luego generar un informe, mientras el código síncrono inspecciona el resultado de cada fase antes de iniciar la siguiente.
Llamar repetidamente a asyncio.run() crea loops independientes. Los recursos, tareas y datos de contexto de uno no se reutilizan naturalmente en el siguiente. Runner encapsula el ciclo de vida y permite varias llamadas secuenciales a run() dentro del mismo entorno.
Uso básico
import asyncio
async def cargar_configuracion():
await asyncio.sleep(0.1)
return {"entorno": "produccion"}
async def sincronizar(config):
await asyncio.sleep(0.1)
return f"sincronizado en {config['entorno']}"
with asyncio.Runner() as runner:
config = runner.run(cargar_configuracion())
resultado = runner.run(sincronizar(config))
print(resultado)El context manager cierra el Runner al terminar el bloque. Cada llamada a run() recibe un awaitable y devuelve su resultado o propaga su excepción. El loop permanece disponible entre llamadas.
No uses Runner dentro de un loop activo
Al igual que asyncio.run(), Runner.run() no puede invocarse cuando ya existe un event loop en ejecución en el mismo thread. Esto es habitual en notebooks, servidores web asíncronos, interfaces gráficas y frameworks que controlan el loop. En esos entornos, usa await directamente o integra la corrutina con el mecanismo del framework.
async def flujo_existente():
config = await cargar_configuracion()
return await sincronizar(config)
# Dentro de código async, usa await.
# No crees un Runner anidado.Esta regla evita loops anidados y planificación impredecible. Una biblioteca reutilizable debería exponer funciones asíncronas y dejar el inicio del loop a la aplicación ejecutable.
Compartir valores de ContextVar
Las variables de contexto son útiles para IDs de correlación, tenants, trazas y metadatos asociados a un trabajo. Runner mantiene un contexto que puede reutilizarse entre llamadas. También puede pasarse un contexto específico a run().
import asyncio
import contextvars
job_id = contextvars.ContextVar("job_id", default="desconocido")
async def registrar_fase(nombre):
print(nombre, job_id.get())
ctx = contextvars.copy_context()
ctx.run(job_id.set, "job-847")
with asyncio.Runner() as runner:
runner.run(registrar_fase("inicio"), context=ctx)
runner.run(registrar_fase("fin"), context=ctx)No guardes secretos innecesarios en variables de contexto. La propagación facilita la observabilidad, pero no constituye una frontera de autorización ni sustituye controles de seguridad.
Modo debug
El parámetro debug=True activa comprobaciones adicionales de asyncio. Puede revelar callbacks lentos, corrutinas creadas pero nunca esperadas y operaciones ejecutadas desde un thread incorrecto.
with asyncio.Runner(debug=True) as runner:
runner.run(cargar_configuracion())Estas comprobaciones son valiosas en desarrollo e integración continua. Añaden coste y pueden producir logs extensos, por lo que en producción conviene habilitarlas de forma controlada y evitar datos sensibles en los registros.
Creación personalizada del loop
El argumento loop_factory controla cómo se crea el event loop. Puede utilizarse para instrumentación, configuración específica de plataforma o una implementación alternativa. La fábrica debe crear y registrar correctamente el loop.
import asyncio
def crear_loop():
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.set_debug(False)
return loop
with asyncio.Runner(loop_factory=crear_loop) as runner:
runner.run(cargar_configuracion())Usa esta opción solo cuando exista una necesidad concreta. Una fábrica incorrecta puede dejar recursos abiertos o entrar en conflicto con bibliotecas que esperan el comportamiento estándar.
Ctrl+C e interrupción controlada
Runner trata KeyboardInterrupt con cuidado. Cuando el usuario pulsa Ctrl+C, la tarea principal se cancela para que la corrutina pueda ejecutar bloques finally y liberar recursos. Si el programa no responde, una segunda interrupción puede finalizarlo de manera más directa.
async def servidor_temporal():
try:
while True:
await asyncio.sleep(1)
finally:
print("liberando recursos")
with asyncio.Runner() as runner:
runner.run(servidor_temporal())No ocultes CancelledError sin una razón deliberada. Realiza la limpieza y normalmente vuelve a propagar la cancelación. Ignorarla puede impedir un cierre predecible.
Cierre de tareas y executors
Al cerrarse, Runner finaliza generadores asíncronos, apaga el executor predeterminado y cierra el loop. Esto ayuda a evitar threads y descriptores filtrados. La aplicación todavía debe gestionar los recursos que creó, como clientes HTTP, pools de base de datos, archivos, colas y directorios temporales.
async def principal():
cliente = crear_cliente()
try:
return await cliente.obtener_datos()
finally:
await cliente.aclose()Prefiere context managers asíncronos cuando la biblioteca los ofrezca. Hacen visible la propiedad del recurso, simplifican las pruebas y mantienen la limpieza cerca de la creación.
Organizar varias fases
Una aplicación de terminal es un buen caso de uso. Cada fase puede devolver valores al código síncrono, mientras conserva el mismo loop y contexto.
async def validar():
return True
async def importar_datos():
return 120
async def crear_informe(total):
print(f"{total} elementos procesados")
with asyncio.Runner() as runner:
if runner.run(validar()):
total = runner.run(importar_datos())
runner.run(crear_informe(total))No conviertas cada función pequeña en una llamada separada al Runner. Si las operaciones forman un flujo asíncrono continuo, crea una corrutina principal y usa await internamente. Las llamadas múltiples son más apropiadas cuando existen fases realmente separadas controladas por código síncrono.
Concurrencia estructurada
Dentro de una corrutina ejecutada por Runner, utiliza asyncio.TaskGroup para coordinar tareas relacionadas. Propaga fallos de manera consistente y cancela tareas hermanas cuando corresponde.
async def descargar(nombre):
await asyncio.sleep(0.1)
return nombre
async def lote():
async with asyncio.TaskGroup() as grupo:
grupo.create_task(descargar("a"))
grupo.create_task(descargar("b"))Evita crear tareas sueltas sin conservar referencias. Una tarea sin supervisión puede fallar sin un informe adecuado o seguir pendiente durante el cierre.
Timeouts y límites de recursos
Las operaciones externas necesitan plazos. Usa asyncio.timeout() alrededor de una unidad de trabajo y gestiona la expiración en una capa capaz de decidir si debe reintentar, informar o abortar.
async def consultar_servicio():
async with asyncio.timeout(5):
return await llamada_remota()Combina plazos con límites de concurrencia, backoff exponencial, idempotencia y circuit breakers cuando sea necesario. Runner administra el ciclo de vida del loop; no protege automáticamente un servicio contra sobrecarga o dependencias inestables.
Gestión de errores
Una excepción de la corrutina principal sale de run() y puede tratarse en el código síncrono. Captura únicamente errores que la capa actual pueda resolver de manera significativa. Registra contexto útil, pero no ocultes fallos de programación detrás de un resultado genérico.
with asyncio.Runner() as runner:
try:
runner.run(sincronizar({"entorno": "produccion"}))
except TimeoutError:
print("la operación excedió el tiempo")Cuando varias tareas fallan dentro de TaskGroup, Python puede producir un grupo de excepciones. Gestiona subgrupos concretos y conserva la información de cada fallo.
Pruebas
Para pruebas unitarias asíncronas, utiliza preferentemente el soporte nativo del framework. Runner resulta especialmente útil al probar una función síncrona que coordina fases asíncronas. Cada prueba debe cerrar su Runner y evitar estado global compartido.
Prueba éxito, excepciones, cancelación, timeout, liberación de recursos y detección de tareas pendientes. Cuando el comportamiento de señales sea importante, crea pruebas de integración específicas en lugar de hacer que todas las pruebas dependan del sistema operativo.
Threads y trabajo bloqueante
Una llamada bloqueante congela el event loop. Mueve funciones bloqueantes cortas a asyncio.to_thread() y usa procesos para trabajo intensivo de CPU. Limita la cantidad de tareas simultáneas para que el executor no se convierta en una cola sin control.
async def leer_archivo_legacy(path):
return await asyncio.to_thread(path.read_text, encoding="utf-8")Cancelar la corrutina que espera no detiene necesariamente el thread de inmediato. Diseña funciones bloqueantes con duración limitada y mecanismos explícitos de cancelación cuando sean necesarios.
Buenas prácticas
- Usa
asyncio.run()para un único punto de entrada sencillo. - Usa Runner cuando el código síncrono necesita varias fases asíncronas.
- Nunca llames Runner dentro de un loop activo.
- Cierra clientes, pools y archivos explícitamente.
- Propaga la cancelación después de limpiar.
- Usa TaskGroup para tareas relacionadas.
- Aplica timeouts y límites de concurrencia.
- Activa debug en desarrollo.
- Evita loop_factory sin una necesidad real.
- Mantén la responsabilidad del loop en la capa de aplicación.
Continúa con las guías de Academify sobre asyncio en Python, TaskGroup en Python, contextvars en Python y timeouts de asyncio.
Conclusión
asyncio.Runner es una herramienta de alto nivel para controlar el ciclo de vida de asyncio en programas síncronos con varias fases asíncronas. Reutiliza un event loop, conserva contexto, coordina interrupciones y cierra la infraestructura subyacente. Sus mejores casos de uso son aplicaciones de terminal y herramientas administrativas con fases distintas. Dentro de un flujo asíncrono continuo, prefiere una corrutina principal, await, concurrencia estructurada y propiedad explícita de recursos.
