Approach: use asyncio for concurrency and aiohttp for async HTTP requests. Use an asyncio.Semaphore to enforce concurrency limit, aiohttp.ClientTimeout for per-request timeout, and retry transient failures (network errors or 5xx) up to 2 retries with exponential backoff. CLI via argparse. Exit 0 if all return 200 else non-zero and print failing endpoints and reasons.python
#!/usr/bin/env python3
import argparse
import asyncio
import sys
from typing import List, Tuple
import aiohttp
import async_timeout
import backoff # optional, see alternatives below
async def fetch(session: aiohttp.ClientSession, url: str, timeout: float, sem: asyncio.Semaphore) -> Tuple[str, int, str]:
async with sem:
attempt = 0
while attempt <= 2:
try:
async with session.get(url, timeout=timeout) as resp:
status = resp.status
if status >= 500 and attempt < 2:
attempt += 1
await asyncio.sleep(0.5 * (2 ** (attempt - 1)))
continue
text = await resp.text()
return url, status, text[:200]
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt < 2:
attempt += 1
await asyncio.sleep(0.5 * (2 ** (attempt - 1)))
continue
return url, -1, str(e)
return url, -1, "unknown"
async def run_checks(urls: List[str], concurrency: int, timeout: float) -> List[Tuple[str, int, str]]:
sem = asyncio.Semaphore(concurrency)
timeout_obj = aiohttp.ClientTimeout(total=None) # per-request handled below
async with aiohttp.ClientSession(timeout=timeout_obj) as session:
tasks = [fetch(session, u, timeout, sem) for u in urls]
return await asyncio.gather(*tasks)
def load_urls(path: str) -> List[str]:
with open(path, 'r') as f:
return [line.strip() for line in f if line.strip() and not line.startswith('#')]
def main():
p = argparse.ArgumentParser(description="Concurrent health-check tool")
p.add_argument("file", help="File with one URL per line")
p.add_argument("--concurrency", "-c", type=int, default=10)
p.add_argument("--timeout", "-t", type=float, default=5.0)
args = p.parse_args()
urls = load_urls(args.file)
if not urls:
print("No URLs provided", file=sys.stderr); sys.exit(2)
results = asyncio.run(run_checks(urls, args.concurrency, args.timeout))
failures = []
for url, status, info in results:
if status != 200:
failures.append((url, status, info))
if failures:
print("Failing endpoints:")
for u, s, info in failures:
print(f"- {u} => status={s} info={info}")
sys.exit(1)
print("All checks OK")
sys.exit(0)
if __name__ == "__main__":
main()
