async_crawler.py en Python

Consultar muchas URLs concurrentemente y reportar tamaños y tiempos.

"""Obtiene varias URLs a la vez e informa tamaños y tiempos."""
import asyncio
import time
import urllib.request

URLS = [
    "https://example.com",
    "https://example.org",
    "https://example.net",
]


def obtener(url):
    with urllib.request.urlopen(url, timeout=10) as res:
        return len(res.read())


async def obtener_todas(urls):
    # las llamadas urllib bloqueantes saltan a hilos; asyncio las arrea
    tareas = [asyncio.to_thread(obtener, url) for url in urls]
    tamanos = await asyncio.gather(*tareas, return_exceptions=True)
    return dict(zip(urls, tamanos))


def principal():
    inicio = time.perf_counter()
    resultados = asyncio.run(obtener_todas(URLS))
    transcurrido = time.perf_counter() - inicio
    for url, tamano in resultados.items():
        if isinstance(tamano, Exception):
            print(f"{url}: falló ({tamano})")
        else:
            print(f"{url}: {tamano:,} bytes")
    print(f"obtuvo {len(resultados)} urls en {transcurrido:.2f}s")


if __name__ == "__main__":
    principal()

Cómo funciona

1. asyncio.to_thread corre el urllib bloqueante en hilos.
2. gather las espera todas con captura de errores.
3. El cronómetro muestra la ganancia de la concurrencia.

Palabras clave y builtins usados aquí

• as
• async
• await
• def
• dict
• else
• for
• if
• isinstance
• len
• print
• return
• with
• zip

El intento, en números

Líneas

38

Caracteres a escribir

934

Tokens

217

Ritmo de tres estrellas

110 tpm

Al ritmo de tres estrellas de 110 tokens por minuto, este intento toma unos 118 segundos.

Escribe este fragmento

Paso 1 de 2 en Bis; paso 52 de 53 en Python pythónico.

← Anterior Siguiente →