0Pricing
Python Academy · Lección

Patrones de E/S asíncrona en producción

Aplique asyncio a clientes HTTP, colas y pipelines del mundo real.

Patrones de E/S asíncrona en producción es una lección gratuita de Python Academy en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de Python Academy, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de Python Academy incluye 4 lecciones en total.

Buenas prácticas del bucle de eventos

Use siempre asyncio.run() como único punto de entrada. Evite mezclar código síncrono y asíncrono sin cuidado y no bloquee nunca el bucle de eventos con tareas de CPU prolongadas.

import asyncio

async def main():
    # Never: time.sleep() — blocks the event loop
    # Always: await asyncio.sleep()
    await asyncio.sleep(1)

asyncio.run(main())

Agrupación de conexiones

Reutilice las conexiones en lugar de abrir una nueva para cada solicitud. Bibliotecas como asyncpg y aiohttp proporcionan pools integrados.

# pip install asyncpg
import asyncpg, asyncio

async def main():
    pool = await asyncpg.create_pool(
        dsn="postgresql://user:pass@host/db",
        min_size=5, max_size=20
    )
    async with pool.acquire() as conn:
        row = await conn.fetchrow("SELECT 1")
    await pool.close()

Productor-consumidor con asyncio.Queue

asyncio.Queue desacopla a los productores y consumidores que se ejecutan como tareas simultáneas.

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        await asyncio.sleep(0.5)
    await q.put(None)  # sentinel

async def consumer(q):
    while (item := await q.get()) is not None:
        print(f"processing {item}")
        q.task_done()

async def main():
    q = asyncio.Queue()
    await asyncio.gather(producer(q), consumer(q))

asyncio.run(main())

Limitación de tasa con Semaphore

Limite las solicitudes salientes simultáneas para evitar sobrecargar las API externas o alcanzar sus límites de tasa.

import asyncio, aiohttp

SEM = asyncio.Semaphore(10)  # 10 concurrent requests max

async def fetch(session, url):
    async with SEM:
        async with session.get(url) as r:
            return await r.json()

async def main(urls):
    async with aiohttp.ClientSession() as s:
        return await asyncio.gather(*[fetch(s, u) for u in urls])

Reintentos con retroceso exponencial

Envuelva las operaciones que se pueden reintentar con una lógica de retroceso para gestionar correctamente los fallos transitorios.

import asyncio

async def retry(coro_fn, retries=3, base=1):
    for attempt in range(retries):
        try:
            return await coro_fn()
        except Exception as e:
            if attempt == retries - 1: raise
            delay = base * 2 ** attempt
            await asyncio.sleep(delay)

Cierre ordenado

Capture asyncio.CancelledError en las tareas de larga duración y gestione las señales del sistema operativo para cerrarlo correctamente.

import asyncio, signal

async def server():
    stop = asyncio.Event()
    loop = asyncio.get_running_loop()
    loop.add_signal_handler(signal.SIGINT, stop.set)
    await stop.wait()
    print("Shutting down gracefully")

Respuestas en streaming

Use generadores asíncronos para transmitir respuestas grandes sin cargarlas por completo en la memoria.

import asyncio, aiohttp

async def stream(url):
    async with aiohttp.ClientSession() as s:
        async with s.get(url) as r:
            async for chunk in r.content.iter_chunked(4096):
                process(chunk)

async def process(chunk):
    await asyncio.sleep(0)
    print(len(chunk), "bytes")

Tareas en segundo plano en FastAPI

Las BackgroundTasks de FastAPI ponen en cola el trabajo para ejecutarlo después de enviar la respuesta, lo que mantiene rápidos los endpoints.

from fastapi import FastAPI, BackgroundTasks

app = FastAPI()

def send_email(to: str, body: str):
    # heavy synchronous work
    pass

@app.post("/register")
async def register(email: str, bg: BackgroundTasks):
    bg.add_task(send_email, email, "Welcome!")
    return {"status": "registered"}

Patrón de heartbeat

Ejecute una tarea periódica en segundo plano junto al bucle principal del servidor mediante create_task.

import asyncio

async def heartbeat():
    while True:
        print("alive")
        await asyncio.sleep(30)

async def main():
    asyncio.create_task(heartbeat())
    await serve_forever()

Perfilado de código asíncrono

Use yappi o py-spy para realizar un perfilado compatible con código asíncrono. Active el modo de depuración de asyncio para recibir advertencias sobre callbacks lentos.

# Enable debug to catch slow callbacks:
# asyncio.run(main(), debug=True)

# PYTHONASYNCIODEBUG=1 python server.py
# logs: Executing <Task coroutine=...> took 0.150 seconds

Cómo evitar errores comunes

Reglas clave: no llame nunca a código bloqueante en el bucle de eventos; espere siempre las corrutinas; cancele las tareas al cerrarse la aplicación; use pools de conexiones; limite la concurrencia con semáforos.

import asyncio

async def bad():
    import time
    time.sleep(2)        # blocks the entire event loop!

async def good():
    await asyncio.sleep(2)  # yields control

Comprobación rápida

¿Qué primitiva de asyncio desacopla a los productores y consumidores que se ejecutan como tareas simultáneas?

Recapitulación

Patrones asíncronos para producción: use pools de conexiones, controle la concurrencia con Semaphore, implemente reintentos con retroceso, transmita datos grandes, cierre correctamente ante las señales y ejecute tareas periódicas con create_task.

Preguntas frecuentes

¿La lección «Patrones de E/S asíncrona en producción» es gratis?

Sí — el texto completo de «Patrones de E/S asíncrona en producción» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de Python Academy, actualiza a CoddyKit PRO. El curso de Python Academy incluye 4 lecciones en total.

¿Qué aprenderé en «Patrones de E/S asíncrona en producción»?

Aplique asyncio a clientes HTTP, colas y pipelines del mundo real. Practicas Python Academy con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.

¿Necesito experiencia previa para empezar Python Academy?

No se requiere experiencia previa. Python Academy en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.

¿Cuánto tiempo toma la lección «Patrones de E/S asíncrona en producción»?

La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.

¿Puedo escribir y ejecutar código en esta lección de Python Academy?

Sí. Cada lección de Python Academy incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.

Todas las lecciones de este curso

  1. Coroutines y el bucle de eventos
  2. await, tareas y recopilación
  3. Gestores e iteradores de contexto asíncronos
  4. Patrones de E/S asíncrona en producción
← Volver a Python Academy