Python Async Programming Patterns

What this prompt does

This prompt produces a comprehensive Python asyncio implementation guide for a [project_type] that must handle a specific [concurrency_requirement]. Rather than abstract theory, it builds the patterns that matter in production: an async application structure with graceful shutdown and TaskGroups, concurrent HTTP with connection pooling and rate limiting, async database access, a producer-consumer pipeline with backpressure, async generators for streaming, observability, and the common pitfalls. Grounding everything in [python_version] and [async_libraries] keeps the code concrete and runnable.

The variables size the system. [pool_size], [rate_limit], and [batch_size] shape the HTTP client's connection pool, token-bucket limiting, and asyncio.gather batching with exponential backoff, while [db_library] and [db_pool_size] define the async database layer and its unit-of-work pattern. [worker_count] and [queue_max_size] drive the producer-consumer pipeline and when backpressure kicks in, [data_source] feeds the streaming async generators, and [executor_type] handles offloading CPU-bound work so it never blocks the event loop. A dead-letter queue captures items that repeatedly fail so one bad payload does not stall the whole pipeline.

When to use it

• Building a service that fans out to many APIs or databases concurrently.
• Implementing rate-limited, pooled HTTP clients with retry and backoff in aiohttp.
• Designing a producer-consumer pipeline with real backpressure and a dead-letter queue.
• Streaming large data sources in chunks with async generators.
• Avoiding event-loop blocking by offloading CPU-bound work via [executor_type].
• Adding async-aware observability like event-loop lag and active task counts.

Example output

You get a structured implementation guide with code: an async main() with event-loop config via [event_loop_config], SIGTERM/SIGINT graceful shutdown, and TaskGroups; an AsyncHTTPClient managing a [pool_size]-connection pool with exponential backoff, a token bucket at [rate_limit], and [batch_size] batching via gather(return_exceptions=True); an async database layer on [db_library] with a [db_pool_size] pool and unit-of-work; a [worker_count]-worker queue pipeline with backpressure at [queue_max_size] and a dead-letter queue; async generators streaming [data_source]; observability helpers; and a pitfalls section covering run_in_executor with [executor_type].

Pro tips

• Use structured concurrency (TaskGroups) so one failing task cancels its siblings cleanly rather than leaking orphaned coroutines.
• Always pass return_exceptions=True to gather when batching [batch_size] requests, then inspect results, so one failure does not sink the whole batch.
• Never run CPU-bound work directly in a coroutine; offload it with run_in_executor and [executor_type], or it blocks the entire event loop.
• Set [queue_max_size] to enforce real backpressure; an unbounded queue just moves your out-of-memory crash downstream.
• Match [pool_size] and [rate_limit] to what the upstream API actually tolerates, not just your peak ambition, or you will trade throughput for a wall of 429 errors.
• Tag async logs with the task name and add timing decorators around key coroutines; without that context, interleaved concurrent logs are nearly impossible to follow when something stalls.