#Un archivo que crece por piezas explicadas
Los bloques de esta página forman comprobador.py en el orden en que aparecen. Cada bloque añade una responsabilidad y no reemplaza los anteriores. La primera página ya explicó el contrato y el servidor local; aquí construiremos el cliente. La última página ejecutará pruebas de fallos y cierre.
El programa conserva resultados pequeños en memoria y los imprime después de cerrar el bucle. Eso evita bloquear las tareas escribiendo continuamente en una terminal lenta. Como admitimos hasta mil entradas, también está acotado el número de resultados. Para millones de registros necesitarías una salida incremental con su propia cola y política de persistencia.
#1. Representar una entrada y su resultado
Una dataclass es una clase orientada a datos: a partir de sus campos genera métodos como el constructor. frozen=True impide reasignar sus atributos normalmente; ayuda a que un worker no cambie la identidad del trabajo que recibió. No convierte de forma profunda cualquier objeto contenido en inmutable; aquí guardamos números y texto.
int | None significa que codigo puede ser un entero o no existir: si la conexión falla antes de recibir cabeceras, no inventaremos un código HTTP. Son anotaciones de tipos, no validadores automáticos. asdict() convertirá un resultado en diccionario para generar JSON.
admitido es tiempo monotónico del bucle. total_ms incluye la espera anterior al worker; cola_ms mide hasta el inicio del worker e incluye una posible espera del productor en put. No son fechas del calendario.
import asyncio
import json
import re
from dataclasses import asdict, dataclass
from urllib.parse import urlsplit
import httpx
@dataclass(frozen=True)
class Trabajo:
numero: int
ruta: str
admitido: float
@dataclass(frozen=True)
class Resultado:
numero: int
ruta: str
estado: str
codigo: int | None
bytes: int
total_ms: float
cola_ms: float
detalle: str = ""
#2. Compartir un ritmo de inicio
La clase Ritmo procede del capítulo de límites. Existe una única instancia por lote y todos los workers pasan por ella. El lock impide conceder dos turnos a la vez. Cada turno efectivo reserva el siguiente instante; si el bucle se retrasó, no compensa disparando una ráfaga.
class Ritmo:
def __init__(self, por_segundo):
self.intervalo = 1 / por_segundo
self.siguiente = 0.0
self.lock = asyncio.Lock()
async def esperar(self):
async with self.lock:
loop = asyncio.get_running_loop()
demora = self.siguiente - loop.time()
if demora > 0:
await asyncio.sleep(demora)
self.siguiente = loop.time() + self.intervalo
Este límite controla inicios locales. Un proveedor puede medir por cuenta y sumar otros procesos; en ese caso debes repartir o coordinar la cuota. Los valores positivos se validan en el dueño del lote, antes de construir Ritmo.
#3. Comprobar una entrada sin cargar todo el cuerpo
comprobar recibe recursos prestados: no crea el cliente ni decide cuándo cerrar el lote. Sí posee su respuesta de streaming. Su deadline se calcula desde la admisión, no desde el inicio de la red. Si ya está vencida al llegar al worker, devuelve deadline sin enviar nada.
El contador http_activos mide intentos dentro del ámbito del cliente, incluida una posible espera de pool; no cuenta conexiones TCP exactas ni trabajo que siga ejecutándose en el servidor después de cancelar el cliente. pico_http es su máximo observado. activos, en el worker, incluirá también la espera del turno de tasa.
aiter_raw(chunk_size=1024) entrega bytes sin descomprimir. Solo retenemos el bloque actual y un contador; no acumulamos el contenido. Al superar el límite cortamos la lectura y el contexto cierra la respuesta. En la prueba de 16 KiB detectaremos el exceso tras 17 KiB: necesitamos observar un bloque adicional para saber que el cuerpo no terminó en el límite.
async def comprobar(cliente, trabajo, ritmo, metricas, plazo, max_bytes):
loop = asyncio.get_running_loop()
inicio = loop.time()
deadline = trabajo.admitido + plazo
codigo, recibidos, detalle = None, 0, ""
estado = "deadline"
try:
if inicio < deadline:
async with asyncio.timeout_at(deadline):
await ritmo.esperar()
metricas["peticiones"] += 1
metricas["http_activos"] += 1
metricas["pico_http"] = max(
metricas["pico_http"], metricas["http_activos"]
)
try:
async with cliente.stream("GET", trabajo.ruta) as respuesta:
codigo = respuesta.status_code
estado = "ok" if 200 <= codigo < 300 else "http_error"
async for bloque in respuesta.aiter_raw(chunk_size=1024):
recibidos += len(bloque)
if recibidos > max_bytes:
estado = "too_large"
break
finally:
metricas["http_activos"] -= 1
except TimeoutError:
estado = "deadline"
except httpx.TimeoutException as error:
estado, detalle = "transport_timeout", type(error).__name__
except httpx.RequestError as error:
estado, detalle = "transport_error", type(error).__name__
return Resultado(
trabajo.numero, trabajo.ruta, estado, codigo, recibidos,
round((loop.time() - trabajo.admitido) * 1000, 2),
round((inicio - trabajo.admitido) * 1000, 2), detalle,
)
El finally devuelve el contador a cero incluso si un timeout cancela la lectura. Los errores de transporte se convierten en resultados. CancelledError no se captura, así que el dueño puede cancelar el grupo. Tampoco atrapamos errores de programación con un except Exception genérico.
Aquí un estado fuera de 2xx es http_error, incluidas las redirecciones. Leemos su cuerpo bajo los mismos límites; si además excede el tamaño, el resultado es too_large y conserva el código recibido. Es una precedencia explícita, no una interpretación accidental.
#4. Crear el lote y admitir solo entradas conocidas
ejecutar es la API asíncrona del programa. Las opciones después de * deben pasarse por nombre, para que plazo=1 no se confunda con lectura=1. La firma y la validación explican qué configura el llamador.
urlsplit separa esquema, host, puerto y otras partes del destino. Restringimos el laboratorio a loopback. La expresión regular acepta exclusivamente las rutas enseñadas y un identificador numérico opcional: \d{1,6} significa entre uno y seis dígitos. fullmatch exige que coincida toda la entrada. No es un filtro general para URLs de Internet.
El iterable rutas puede ser un generador sencillo que produzca una ruta cada vez. Debe avanzar sin hacer I/O bloqueante y entregar cadenas pequeñas; si su origen es una base o archivo grande, adapta esa lectura como una fase propia. El límite de mil entradas también limita la lista final. Una entrada inválida avanzada dentro de un generador puede detectarse después de consultar otras: la validación incremental no convierte el lote en una transacción.
async def ejecutar(rutas, *, base_url="http://127.0.0.1:8767",
workers=3, capacidad=6, plazo=1.0, plazo_lote=5.0,
por_segundo=20.0, max_bytes=16_384, lectura=0.2):
if any(valor <= 0 for valor in (
workers, capacidad, plazo, plazo_lote, por_segundo, max_bytes, lectura
)):
raise ValueError("los límites deben ser positivos")
destino = urlsplit(base_url)
if (destino.scheme != "http" or destino.hostname != "127.0.0.1"
or destino.port is None or destino.username is not None
or destino.password is not None or destino.path not in {"", "/"}
or destino.query or destino.fragment):
raise ValueError("este laboratorio solo admite HTTP en 127.0.0.1:puerto")
cola = asyncio.Queue(maxsize=capacidad)
ritmo = Ritmo(por_segundo)
resultados = []
metricas = dict(admitidos=0, completados=0, activos=0, pico_activos=0,
pico_cola=0, http_activos=0, pico_http=0, peticiones=0)
tareas = []
async def producir():
for numero, ruta in enumerate(rutas):
if numero >= 1000:
raise ValueError("el lote admite como máximo 1000 entradas")
if not isinstance(ruta, str) or not re.fullmatch(
r"/(ok|slow|error|large|trickle)(\?i=\d{1,6})?", ruta
):
raise ValueError("ruta fuera del laboratorio")
trabajo = Trabajo(numero, ruta, asyncio.get_running_loop().time())
await cola.put(trabajo)
metricas["admitidos"] += 1
metricas["pico_cola"] = max(metricas["pico_cola"], cola.qsize())
for _ in range(workers):
await cola.put(None)
La marca temporal se toma antes de put: un productor que espera porque la cola está llena no regala tiempo extra a esa entrada. admitidos cuenta solo los elementos que sí entraron en la cola. Puede existir un elemento adicional en manos del productor esperando espacio. En una interrupción, ese elemento no se presenta como admitido.
Los centinelas None indican final normal y no se cuentan como trabajos de negocio. Se publican después de todas las entradas, uno por worker. Si la entrada falla, el grupo cancela el conjunto en lugar de esperar centinelas que nunca llegarán.
#5. Mantener un número fijo de workers
Cada worker retira una entrada, ejecuta la comprobación y añade su resultado. No hay un create_task por entrada. Como las actualizaciones del contador y append no contienen suspensiones, las tareas del mismo bucle no se intercalan dentro de esas instrucciones.
async def consumir(numero):
while True:
trabajo = await cola.get()
try:
if trabajo is None:
return
metricas["activos"] += 1
metricas["pico_activos"] = max(
metricas["pico_activos"], metricas["activos"]
)
try:
resultado = await comprobar(
cliente, trabajo, ritmo, metricas, plazo, max_bytes
)
resultados.append(asdict(resultado))
metricas["completados"] += 1
finally:
metricas["activos"] -= 1
finally:
cola.task_done()
task_done corresponde exactamente a cada get, también al centinela. El finally interior libera el contador de actividad aunque la comprobación sea cancelada. Los trabajos cancelados no fabrican un resultado ok ni aumentan completados.
#6. Poner cliente y tareas bajo un dueño común
El cliente engloba al grupo: primero terminan los workers y después se cierran sus conexiones. El deadline de lote engloba también productor y join. Si vence, TaskGroup cancela sus tareas y espera su salida antes de que capturemos TimeoutError.
Los nombres async-productor y async-worker-N ayudan a identificar tareas al diagnosticar. Guardamos las referencias para comprobar su finalización. cliente.is_closed comprueba el estado del cliente una vez cerrado, no por intuición.
cliente = httpx.AsyncClient(
base_url=base_url, trust_env=False, follow_redirects=False,
headers={"Accept-Encoding": "identity"},
limits=httpx.Limits(max_connections=workers, max_keepalive_connections=workers),
timeout=httpx.Timeout(connect=0.2, read=lectura, write=0.2, pool=0.2),
)
motivo = "completo"
async with cliente:
try:
async with asyncio.timeout(plazo_lote):
async with asyncio.TaskGroup() as grupo:
tareas.append(grupo.create_task(producir(), name="async-productor"))
for numero in range(workers):
tareas.append(grupo.create_task(
consumir(numero), name=f"async-worker-{numero}"
))
await cola.join()
except TimeoutError:
motivo = "plazo_lote"
metricas["cliente_cerrado"] = cliente.is_closed
metricas["tareas_terminadas"] = all(tarea.done() for tarea in tareas)
metricas["sin_resultado"] = metricas["admitidos"] - metricas["completados"]
return dict(motivo=motivo, metricas=metricas,
resultados=sorted(resultados, key=lambda item: item["numero"]))
Los números de línea de los bloques no son relevantes: lo importante es conservar la indentación. Este último bloque sigue dentro de ejecutar hasta su return. sin_resultado diferencia entradas admitidas de resultados terminados; puede incluir entradas aún en cola y otras interrumpidas. No afirma que todas llegaron al servidor.
#7. Ejecutarlo y producir el informe
async def main():
rutas = ["/ok", "/error", "/slow", "/large", "/ok?i=2"]
return await ejecutar(rutas)
if __name__ == "__main__":
informe = asyncio.run(main())
print(json.dumps(informe, ensure_ascii=False, indent=2))
Con el servidor de la página anterior en marcha:
python comprobador.py
La lista conserva el orden de entrada, aunque los resultados se obtuvieron en otro orden. Debes encontrar ok, http_error, transport_timeout, too_large y otro ok. El detalle del timeout de /slow será ReadTimeout. Los tiempos variarán con la máquina; los estados y límites son lo que vamos a comprobar.
El informe debe mostrar cliente_cerrado: true, tareas_terminadas: true y actividad final cero. Si el lote agota su plazo, también habrá motivo: "plazo_lote" y un número explícito de admitidos sin resultado. No hemos añadido reintentos para que cada entrada genere como máximo una petición; la política del capítulo anterior permite ampliarlo sabiendo qué cambiaría.
#Qué garantías acabamos de construir
La cantidad de tareas depende de workers, no del tamaño del iterable. La cola frena al productor. Los plazos incluyen las esperas elegidas. El cuerpo no se guarda completo. Cada recurso tiene un ámbito de cierre. Un fallo HTTP individual conserva resultados útiles; un defecto inesperado sigue siendo visible.
Quedan límites deliberados: no hay almacenamiento durable, monitorización periódica, autenticación de servicios ni destinos de Internet. El siguiente capítulo prueba las garantías reales antes de decidir qué ampliación merece la pena.
#Referencias
- dataclasses, expresiones regulares y urlsplit, Python 3.12.
- API de HTTPX,
AsyncClient, límites y respuestas.