Proyecto · Contrato y laboratorio HTTP

Construiremos un comprobador que recibe rutas de servicios administrados, realiza consultas HTTP y produce un informe JSON. Conservará los resultados individuales aunque algunos endpoints fallen. Tendrá un límite de trabajo pendiente, otro de operaciones simultáneas, un ritmo de inicio, un presupuesto por entrada y otro para todo el lote.

Las tres páginas del proyecto incluyen todo el código y su explicación. Para ejecutarlo puedes crear tres archivos en una misma carpeta de trabajo: laboratorio.py, comprobador.py y pruebas.py. Los nombres sirven para que los imports encajen; no necesitas descargar un proyecto ni buscar piezas fuera de estas páginas.

El comprobador no interpreta un 503 como un defecto de Python. Devuelve un resultado http_error y continúa con las demás entradas. Un error de programación, en cambio, detiene el grupo. Un plazo de lote agotado devuelve solo los resultados completos y un contador explícito de admitidos sin resultado.

No son valores recomendados para cualquier proveedor. Son límites pequeños que podemos observar en el laboratorio. Cambiarlos después de medir es parte del diseño.

Ampliar diagrama ↗

El informe del lote tiene además un motivo: completo o plazo_lote. «Completo» significa que procesó cada entrada admitida, no que todas devolvieron ok. Una cancelación externa se propaga al llamador después de limpiar; no se disfraza de informe completo.

La prueba debe producir lentitud y errores cuando nosotros decidamos. Usaremos http.server, un módulo de la biblioteca estándar, únicamente en 127.0.0.1. No necesita framework ni credenciales y no consulta servicios ajenos.

ThreadingHTTPServer atiende conexiones mediante hilos. Eso permite que una respuesta lenta no detenga todas las demás. La clase Respuesta hereda el mecanismo que interpreta HTTP y define do_GET, el método que se llama cuando llega una petición GET. No estamos enseñando a escribir un servidor web de producción: este componente existe para controlar el experimento.

El flujo de /trickle permite distinguir timeout de lectura y deadline total. Los datos llegan con frecuencia, pero la respuesta completa sigue tardando. ?i=123 identifica entradas distintas sin cambiar el comportamiento de la ruta.

La instancia Laboratorio posee un lock de hilos y dos mediciones: peticiones recibidas y puertos de cliente observados. Varias peticiones desde el mismo puerto cliente, dentro de esta prueba y conexión TCP, permiten observar reutilización. Esa medición no es un contador universal de conexiones para producción: los puertos pueden reutilizarse con el tiempo.

protocol_version = "HTTP/1.1" y Content-Length permiten mantener conexiones sin confundir dónde termina una respuesta. time.sleep es intencional aquí: se ejecuta en el hilo de una conexión del servidor, no en el event loop del comprobador. El cuerpo son bytes y wfile.write los envía al cliente.

Si el cliente cancela una respuesta, el servidor puede descubrirlo al escribir. Capturamos únicamente BrokenPipeError y ConnectionResetError, errores esperados para esa prueba. No ocultamos cualquier fallo del servidor.

import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlsplit


class Laboratorio(ThreadingHTTPServer):
    daemon_threads = True

    def __init__(self, direccion):
        super().__init__(direccion, Respuesta)
        self.lock = threading.Lock()
        self.puertos_cliente = set()
        self.peticiones = 0


class Respuesta(BaseHTTPRequestHandler):
    protocol_version = "HTTP/1.1"

    def log_message(self, formato, *args):
        pass

    def do_GET(self):
        with self.server.lock:
            self.server.puertos_cliente.add(self.client_address[1])
            self.server.peticiones += 1
        ruta = urlsplit(self.path).path
        codigo = 503 if ruta == "/error" else 200
        if ruta == "/large":
            cuerpo = b"x" * 32_768
        elif ruta == "/trickle":
            cuerpo = b"x" * 4096
        elif ruta in {"/ok", "/slow", "/error"}:
            cuerpo = b'{"estado":"respuesta local"}'
        else:
            codigo, cuerpo = 404, b"ruta desconocida"
        if ruta == "/slow":
            time.sleep(0.6)
        elif ruta == "/ok":
            time.sleep(0.03)
        try:
            self.send_response(codigo)
            self.send_header("Content-Type", "application/octet-stream")
            self.send_header("Content-Length", str(len(cuerpo)))
            self.end_headers()
            if ruta == "/trickle":
                for inicio in range(0, len(cuerpo), 1024):
                    self.wfile.write(cuerpo[inicio:inicio + 1024])
                    self.wfile.flush()
                    time.sleep(0.08)
            else:
                self.wfile.write(cuerpo)
        except (BrokenPipeError, ConnectionResetError):
            pass  # El cliente puede haber agotado su plazo y cerrado.


if __name__ == "__main__":
    with Laboratorio(("127.0.0.1", 8767)) as servidor:
        print("Laboratorio listo en http://127.0.0.1:8767", flush=True)
        try:
            servidor.serve_forever(poll_interval=0.05)
        except KeyboardInterrupt:
            pass

Los hilos de este servidor de prueba son daemon: no prolongan indefinidamente el cierre del proceso. Sus demoras están acotadas y no hacen escrituras de negocio. No extrapoles esa decisión a un worker que deba terminar una transacción o conservar un archivo.

Activa el entorno virtual explicado en la portada y ejecuta:

python laboratorio.py

Debes ver Laboratorio listo en http://127.0.0.1:8767. Si el puerto está ocupado, detén tu instancia anterior o cambia ese puerto y el base_url del comprobador de forma coherente. Las pruebas automáticas de la tercera página elegirán un puerto libre sin depender del 8767.

Desde otra terminal puedes inspeccionar una respuesta:

curl -i http://127.0.0.1:8767/error

Debe aparecer 503 Service Unavailable. curl es un cliente de terminal para inspección; el programa final no depende de él. Si falla la conexión, el cliente todavía no está probando concurrencia: primero comprueba proceso, dirección y puerto.

El motor de la siguiente página solo admite http://127.0.0.1:puerto y cinco rutas conocidas. Es una decisión para hacer reproducible el laboratorio. No acepta una lista arbitraria de URLs proporcionadas por usuarios.

Para convertirlo en una herramienta de tu equipo, configura nombres de servicios y destinos permitidos administrativamente, habilita HTTPS con verificación y define límites por destino. Si aceptas URLs ajenas, necesitarás una política completa de resolución DNS, redes permitidas, puertos, redirecciones y salida de red. Cambiar una comprobación de texto no resuelve todos esos problemas.

En la siguiente página construiremos cada pieza del motor y las uniremos bajo el mismo dueño. El servidor puede quedarse abierto mientras lo pruebas.

• http.server y threading, documentación de Python 3.12.