Puente Síncrono/Asíncrono
Las bases de código reales mezclan bibliotecas bloqueantes con frameworks asíncronos. Utiliza el puente con asyncio.to_thread (3.9+), loop.run_in_executor, y asyncio.run solo en los puntos de entrada; nunca llames a asyncio.run desde dentro de un bucle en ejecución.
Receta
import asyncio
def blocking_io() -> str:
return "done"
async def main() -> None:
result = await asyncio.to_thread(blocking_io)
print(result)Cuándo usar esto:
- SDK de base de datos/HTTP síncrono heredado en una ruta de FastAPI
- Llamadas de archivo o criptografía bloqueantes con uso ligero de CPU
- Invocar asíncrono desde un wrapper CLI síncrono (con cuidado)
- Migración gradual de Flask a asyncio
- Biblioteca de terceros sin soporte asíncrono
Ejemplo Funcional
import asyncio
import time
from concurrent.futures import ThreadPoolExecutor
def slow_sync(n: int) -> int:
time.sleep(0.05)
return n * 2
async def via_to_thread() -> int:
return await asyncio.to_thread(slow_sync, 21)
async def via_executor() -> int:
loop = asyncio.get_running_loop()
with ThreadPoolExecutor(max_workers=2) as pool:
return await loop.run_in_executor(pool, slow_sync, 10)
async def main() -> None:
print(await via_to_thread())
print(await via_executor())
asyncio.run(main())Lo que esto demuestra:
to_threadprograma la llamada en el pool de hilos predeterminadorun_in_executoracepta un ejecutor personalizado para aislamiento- El trabajo bloqueante se ejecuta fuera del hilo del bucle de eventos
- El bucle permanece receptivo para otras corrutinas
Análisis Profundo
Matriz de Direcciones
| De -> A | Herramienta |
|---|---|
| bloqueante síncrono -> asíncrono | to_thread / ejecutor |
| entrada síncrona -> asíncrona | asyncio.run(main()) una vez |
| intermedio síncrono -> asíncrono | asyncio.run_coroutine_threadsafe (avanzado) |
Vinculado a CPU
- Prefiere
ProcessPoolExecutorpara Python con uso intensivo de CPU to_threadno evita el GIL para el cómputo
Trampas Comunes
asyncio.rundentro de un bucle en ejecución -RuntimeError. Solución: hacerawaitde la corrutina directamente o usar el puente seguro para hilos.- Inundación de
to_thread- miles de hilos en cola. Solución: semáforo + pool limitado. - ORM bloqueante en cada solicitud - el pool de hilos se convierte en un cuello de botella. Solución: controlador asíncrono nativo.
- Compartir sesiones de ORM entre hilos - inseguro. Solución: sesión por hilo/tarea.
asyncio.runanidado en pruebas - usa el fixture de bucle único depytest-asyncio.
Alternativas
| Alternativa | Usar Cuando | No Usar Cuando |
|---|---|---|
| Cliente asíncrono nativo | Disponible para la dependencia | N/A |
| Framework síncrono (WSGI) | Toda la aplicación bloqueante | Necesidad de E/S concurrente alta |
| anyio | Abstracción asíncrona portable | Política de solo biblioteca estándar |
Preguntas Frecuentes
¿`to_thread` vs `run_in_executor`?
to_thread usa el ejecutor predeterminado - más simple; ejecutor personalizado cuando se necesita aislamiento.
¿Cuántos workers de hilo?
Por defecto ~ min(32, cpu+4) - menor si las tareas se bloquean mucho para evitar acumulación.
¿Llamar a asíncrono desde REPL síncrono?
asyncio.run(expr()) solo en una shell nueva - no desde un Jupyter asíncrono sin nest_asyncio (evitar en producción).
¿`def` vs `async def` en FastAPI?
async def para await; def síncrono se ejecuta automáticamente en el pool de hilos - aún así prefiere E/S asíncrona real.
¿Pool de procesos desde asíncrono?
await loop.run_in_executor(process_pool, fn) - vigila la serialización y la guarda principal.
¿`contextvars` en hilos?
contextvars.copy_context().run para propagar el ID de solicitud al trabajo de to_thread.
¿Vistas asíncronas de Django?
El ORM sigue siendo en gran medida síncrono - wrapper sync_to_async con indicadores de sensibilidad de hilo.
¿Cuándo no usar el puente?
Si >30% del tiempo de ruta se pasa en el ejecutor - migrar a una biblioteca asíncrona o a un modelo de despliegue de workers síncronos.
¿`time.sleep` en un hilo está bien?
Sí, en un worker de to_thread; nunca en el cuerpo de una corrutina.
¿Probar puentes?
Simular slow_sync para incrementar un contador; afirmar que el bucle responde a través de una tarea de latido concurrente.
Relacionado
- threading - primitivas de hilos
- concurrent.futures - ejecutores
- Debugging Async Code - detección de bloqueos
- Async Best Practices - cuándo migrar del puente
Versiones de la pila: Esta página fue escrita 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+.