Skip to content
CloudOps
Newsletter
All prompts
AI for Bash & Python Automation Difficulty: Advanced ClaudeChatGPT

Async Concurrent HTTP Poller with asyncio and httpx Prompt

Build a fast, bounded-concurrency async poller/fetcher that hits many endpoints with asyncio and httpx, with rate limiting, retries, timeouts, and structured results.

Target user
Python engineers fetching or polling many HTTP endpoints concurrently
Difficulty
Advanced
Tools
Claude, ChatGPT

The prompt

You are a senior Python engineer who writes high-throughput async I/O clients that stay polite to upstream services and never leak connections.

I will provide:
- The list/source of URLs (static list, paginated API, or a queue)
- Auth requirements (bearer, mTLS, query keys)
- Upstream rate limits and SLA expectations
- What "done" looks like (one-shot fetch vs continuous poll)

Produce a single, runnable module using `asyncio` + `httpx.AsyncClient`. Requirements:

1. **Bounded concurrency** — use an `asyncio.Semaphore` (configurable, default 10), never an unbounded `gather` over thousands of tasks. Reuse ONE `AsyncClient` with an explicit `limits=httpx.Limits(...)` connection pool.

2. **Timeouts** — set per-request `httpx.Timeout` (connect, read, write, pool). No request may hang forever.

3. **Retries with backoff** — retry only on 429, 5xx, and transport errors. Honor `Retry-After`. Exponential backoff with full jitter, capped attempts. Do NOT retry 4xx (except 429).

4. **Rate limiting** — token-bucket or simple delay to respect upstream RPS. Make it a reusable async limiter, not sleeps scattered in the code.

5. **Structured results** — return a dataclass per URL: url, status, elapsed_ms, attempt_count, ok flag, and either parsed body or error. Never let one failure abort the batch; gather with `return_exceptions` or per-task try/except.

6. **Polling mode** — when continuous, poll on an interval with graceful shutdown on SIGINT/SIGTERM (cancel tasks, drain, close client).

7. **Observability** — structured log lines (JSON optional) and a summary: total, ok, failed, p50/p95 latency.

8. **Backpressure** — if the source is a stream/queue, use an `asyncio.Queue` with worker tasks rather than loading everything into memory.

Output: (a) the full module with type hints, (b) a `main()` guarded by `if __name__ == "__main__"`, (c) a tiny `pytest` test using `respx`/`httpx.MockTransport`, (d) notes on tuning concurrency vs connection-pool size.

Bias toward: one shared client, explicit timeouts everywhere, idempotent retries only, clean cancellation. Avoid `requests` in async code and avoid `asyncio.run` inside libraries.
Newsletter

Free: the DevOps AI Incident-Triage Cheat Sheet

Subscribe and we’ll send you the one-page cheat sheet — plus weekly AI prompts, automation ideas, and tool reviews for infrastructure engineers. One email a week. No spam, unsubscribe anytime.

  • AI Incident-Triage Cheat Sheet (PDF)
  • Access to 1,603 DevOps AI prompts
  • One practical workflow email per week