Build async generator pipelines for streaming data processing with backpressure, error handling, and graceful shutdown.
Act as a Python 3.13 async expert. Generate an async generator pipeline for high-throughput streaming data processing. ## Input Parameters - **Data Source**: [DATA_SOURCE] (API, file, queue, websocket) - **Pipeline Stages**: [STAGE_DESCRIPTIONS] - **Concurrency Limit**: [MAX_CONCURRENT] - **Error Strategy**: [ERROR_HANDLING] ## Instructions 1. Create an async generator for the data source with proper resource cleanup (async with). 2. Chain async generator stages using `async for` with transformation logic. 3. Implement backpressure using asyncio.Semaphore. 4. Add TaskGroup for concurrent processing within stages. 5. Handle errors per-item (skip, retry, dead-letter) without killing the pipeline. 6. Implement graceful shutdown on SIGTERM using asyncio.Event. 7. Add structured logging with stage timing metrics. ## Output Return the complete pipeline module, each stage as a separate async generator function, and an orchestrator that wires them together.
Free to copy and use. Compatible with Claude 4 Opus, GPT-5, Gemini 2.0 Flash.
Specify the data source type and connection details. Describe each pipeline stage (e.g., parse, validate, enrich, store). Set concurrency limit and error handling strategy.
Initial release
Sign in and download this prompt to leave a review.