Here is the fully translated and refined English article, tailored for a technical audience on Dev.to. All technical details, structural elements, and the complete code block have been preserved and translated into professional engineering terminology. The conclusion has been distilled to focus purely on the technical takeaways.
To achieve a script that is highly portable, runs immediately in any environment without prior setup, and minimizes dependencies, we adopted the following architectural design:
urllib.request for HTTP communications, alongside json and difflib. ProcessPoolExecutor. as_completed loop, applying limits via future.result(timeout=remaining).
Below is the complete code that was evaluated during our testing phase—a script that ultimately succumbed to deadlocks and cascading timeouts.
import sys
import json
import urllib.request
import urllib.error
from concurrent.futures import ProcessPoolExecutor, as_completed
import time
import difflib
def call_endpoint(endpoint_info):
"""
Sends an HTTP request to a single endpoint and returns the result.
Placed at the module top-level to allow serialization by the process pool.
"""
name = endpoint_info.get("name", "Unknown")
url = endpoint_info.get("url")
headers = endpoint_info.get("headers", {})
payload = endpoint_info.get("payload", {})
timeout = endpoint_info.get("timeout", 10)
start_time = time.time()
try:
data = json.dumps(payload).encode("utf-8")
req = urllib.request.Request(url, data=data, headers=headers, method="POST")
with urllib.request.urlopen(req, timeout=timeout) as response:
elapsed = time.time() - start_time
body = response.read().decode("utf-8")
try:
parsed_body = json.loads(body)
except json.JSONDecodeError:
parsed_body = body
return {
"name": name,
"status": "success",
"status_code": response.status,
"elapsed": round(elapsed, 3),
"response": parsed_body
}
except urllib.error.HTTPError as e:
elapsed = time.time() - start_time
err_body = e.read().decode("utf-8", errors="ignore")
return {
"name": name,
"status": "http_error",
"status_code": e.code,
"elapsed": round(elapsed, 3),
"error": err_body
}
except urllib.error.URLError as e:
elapsed = time.time() - start_time
return {
"name": name,
"status": "url_error",
"status_code": None,
"elapsed": round(elapsed, 3),
"error": str(e.reason)
}
except Exception as e:
elapsed = time.time() - start_time
return {
"name": name,
"status": "timeout_or_unknown",
"status_code": None,
"elapsed": round(elapsed, 3),
"error": str(e)
}
def generate_markdown_report(results, test_case_name):
report = []
report.append(f"# AI Validation & Diff Report: {test_case_name}")
report.append("\n## 1. Execution Summary\n")
report.append(
"| Endpoint | Status | HTTP Code | Latency (s) | Error Details |\n"
"| :--- | :--- | :--- | :--- | :--- |"
)
success_responses = {}
for r in results:
name = r["name"]
status = r["status"]
code = r["status_code"] if r["status_code"] is not None else "-"
elapsed = r["elapsed"]
err = "-"
if status == "success":
success_responses[name] = r["response"]
else:
err = r.get("error", "Unknown error").replace("\n", " ")
report.append(f"| {name} | {status} | {code} | {elapsed} | {err} |")
report.append("\n## 2. Response Outputs\n")
for r in results:
report.append(f"### [{r['name']}] Output")
report.append("```
json")
report.append(json.dumps(r.get("response") or r.get("error"), ensure_ascii=False, indent=2))
report.append("
```\n")
report.append("## 3. Structural & Textual Diff Analysis\n")
names = list(success_responses.keys())
if len(names) < 2:
report.append("Skipping diff analysis because fewer than 2 successful responses were received.\n")
else:
for i in range(len(names)):
for j in range(i + 1, len(names)):
n1, n2 = names[i], names[j]
text1 = json.dumps(success_responses[n1], ensure_ascii=False, indent=2).splitlines()
text2 = json.dumps(success_responses[n2], ensure_ascii=False, indent=2).splitlines()
diff = list(difflib.unified_diff(text1, text2, fromfile=n1, tofile=n2, lineterm=""))
report.append(f"### Diff: {n1} vs {n2}")
if diff:
report.append("```
diff")
report.extend(diff[:50])
if len(diff) > 50:
report.append("... (diff truncated)")
report.append("
```\n")
else:
report.append("No differences found (Exact match).\n")
return "\n".join(report)
def main():
input_data = sys.stdin.read()
if not input_data.strip():
print("Error: Test cases must be provided via standard input.", file=sys.stderr)
sys.exit(1)
try:
test_suite = json.loads(input_data)
except json.JSONDecodeError as e:
print(f"Error: Failed to parse JSON: {e}", file=sys.stderr)
sys.exit(1)
test_name = test_suite.get("test_name", "Unnamed Test")
endpoints = test_suite.get("endpoints", [])
if not endpoints:
print("Error: No target endpoints defined in the test suite.", file=sys.stderr)
sys.exit(1)
results = []
max_workers = min(len(endpoints), 16)
global_timeout = 10.0
start_time = time.time()
with ProcessPoolExecutor(max_workers=max_workers) as executor:
future_to_endpoint = {executor.submit(call_endpoint, ep): ep for ep in endpoints}
for future in as_completed(future_to_endpoint):
ep = future_to_endpoint[future]
name = ep.get("name", "Unknown")
elapsed_total = time.time() - start_time
remaining = max(0.1, global_timeout - elapsed_total)
try:
res = future.result(timeout=remaining)
results.append(res)
except Exception as e:
results.append({
"name": name,
"status": "timeout_or_error",
"status_code": None,
"elapsed": round(time.time() - start_time, 3),
"error": str(e) or "Execution timed out or failed"
})
markdown_report = generate_markdown_report(results, test_name)
print(markdown_report)
if __name__ == "__main__":
main()
💡 For immediate deployment: The complete source code suite (ZIP) for this architecture is available on Gumroad for $0+ (Pay What You Want).
After repeating our QA test suite three times, the precise mechanisms that dragged this system into fatal deadlocks and latency traps became evident.
While multiprocessing is highly effective for CPU-intensive tasks, it is fundamentally unsuitable for network I/O-heavy operations like calling external LLM APIs.
The overhead introduced by Inter-Process Communication (IPC), object serialization/deserialization, and OS-level process spawning is not negligible. When multiple external APIs simultaneously experienced delayed responses, the aggressive context switching between processes severely bottlenecked system resources.
To enforce the strict 10-second global limitation, we integrated the following calculation into the core loop:
elapsed_total = time.time() - start_time
remaining = max(0.1, global_timeout - elapsed_total)
res = future.result(timeout=remaining)
The fatal flaw here is that as_completed yields futures in the order they complete.
If the interpreter first evaluates a future from an endpoint that experienced heavy latency (e.g., a local LLM taking 8 seconds to respond), the remaining time shrinks drastically. Consequently, when the loop processes the next future—even one from an API that responded almost instantaneously—it applies the severely depleted remaining time (e.g., 0.1 seconds). This triggers a cascading timeout collapse, forcefully throwing TimeoutExpired exceptions and terminating requests that actually succeeded at the network layer.
asyncio)
Driven by an absolute constraint to maintain "zero third-party dependencies," we attempted to forcefully parallelize the blocking urllib.request using multi-processing, rather than adopting Python's native asyncio combined with a non-blocking wrapper (like http.client wrapped in an executor) or leveraging asynchronous runtimes.
By abandoning efficient event-loop multiplexing (which threads or async coroutines handle natively), we constructed a fragile, synchronous wait-state architecture wholly dependent on process scaling—a fundamentally broken approach for scaling HTTP concurrency.
Based on the failures observed in this prototype, we offer the following critical insights for engineers designing similar asynchronous validation pipelines:
httpx or native timeout parameter in asyncio.wait()), ensuring deterministic behavior regardless of the order of completion.
A design that appeared conceptually sound on paper was ultimately shattered against the 10-second barrier by real-world network jitter and a fundamental misapplication of the process concurrency model. However, highlighting the exact boundary limits of standard library concurrency models serves as a solid foundation for more robust architectural decisions in the future.
In software engineering, the structural understanding of why failing code breaks is the most valuable asset for future success. We hope this postmortem acts as a navigational compass for developers stepping into the demanding territory of highly concurrent API validation.
If this engineering log saved your production server (and your sanity), consider supporting our architecture on GitHub Sponsors.