Async Pipeline Backpressure

Introduction

Welcome back to Building an Async CLI Tool for ETL Pipelines in Python! In the previous three lessons, we constructed a self-validating domain model with frozen dataclasses and descriptors, built streaming parsers that normalize CSV and JSON Lines data into validated transactions, and implemented a declarative router using structural pattern matching to dispatch transactions to appropriate business handlers. Today's focus is Async Pipeline with Backpressure : transforming our sequential processor into a concurrent, resilient system that can handle multiple data streams efficiently while maintaining control over resource usage. We'll construct an asynchronous engine that combines producers, consumers, and a bounded queue, leveraging asyncio.TaskGroup for structured concurrency. The key insight is that a small queue naturally creates backpressure, preventing producers from overwhelming consumers and keeping memory usage predictable. By the end of this lesson, we'll have implemented a complete async pipeline that streams parsed records through concurrent workers, collects both successful results and validation errors, and coordinates graceful shutdown across all tasks. The architecture scales naturally; you can adjust the number of workers and queue size to match your workload characteristics without restructuring the code. Let's begin by understanding why backpressure matters in async pipelines.

The Producer-Consumer Pattern with Backpressure

Asynchronous pipelines excel at maximizing throughput when I/O operations dominate the workload. Instead of waiting for each file read or network request to complete before starting the next, we can initiate many operations concurrently and process results as they arrive. However, this concurrency introduces a new challenge: what happens when producers generate data faster than consumers can process it? Without backpressure, the producer would flood memory with unprocessed items, eventually exhausting available resources and crashing the application. The solution is a bounded queue : a queue with a maximum size that blocks producers when full. When the queue reaches capacity, the producer's await queue.put(item) suspends until a consumer removes an item, freeing space. This creates natural backpressure that adapts to the consumer's processing speed without explicit rate limiting or complex coordination logic. The pattern works because asyncio's cooperative multitasking ensures fairness: when the producer blocks on a full queue, other tasks like consumers continue running, processing items and emptying the queue. Once space becomes available, the producer resumes automatically. This self-regulating behavior makes the system resilient; if consumers slow down due to downstream latency, the producer automatically adjusts its pace.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Orchestrating with TaskGroup

The run_pipeline function orchestrates the entire async pipeline, managing the queue, tasks, and file handling:

Python

async def run_pipeline(
    path: Path, kind: str, *, workers: int = 3, maxsize: int = 3
) -> Tuple[List[Dict[str, Any]], List[str]]:
    if workers < 1:
        workers = 1
    q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
    results: List[Dict[str, Any]] = []
    errors: List[str] = []

    if kind == "csv":
        fh = path.open("r", encoding="utf-8", newline="")
        parse = lio.parse_csv
    elif kind == "jsonl":
        fh = path.open("r", encoding="utf-8")
        parse = lio.parse_jsonl
    else:
        raise ValueError("kind must be 'csv' or 'jsonl'")

The function accepts a file path, format kind, and tuning parameters for workers and queue size. We validate that at least one worker exists, create a bounded queue with the specified maxsize, and initialize empty result and error lists that will be shared across all tasks.

The file handling logic opens the appropriate file handle and selects the matching parser function. Note that we open the file synchronously here; the async pipeline consumes the parsed records, but the parsing itself remains synchronous because file I/O in Python's standard library isn't truly async. The async benefits come from concurrent processing of multiple transactions, not from async file reads.

Python

    try:
        async with asyncio.TaskGroup() as tg:
            raw_records = parse(fh)
            tg.create_task(_producer(q, raw_records, errors, workers))
            for _ in range(workers):
                tg.create_task(_consumer(q, results))
            await q.join()
    finally:
        fh.close()
    return results, errors

The async with asyncio.TaskGroup() context manager provides structured concurrency: all tasks created within the block are automatically awaited when the block exits, and if any task raises an exception, all other tasks are canceled. This guarantees cleanup and prevents orphaned tasks.

We create the producer task first, passing it the parsed records generator, the queue, the errors list, and the consumer count. Then we create multiple consumer tasks in a loop, each sharing the same queue and results list. The producer and consumers run concurrently; as soon as we call create_task, the task begins executing alongside other tasks.

The await q.join() is crucial: it blocks until the queue's internal counter of unfinished tasks reaches zero, meaning all items that were put have had task_done called. This ensures we don't exit the TaskGroup until all consumers have processed all transactions and sentinels. The finally block guarantees file closure even if an exception occurs during processing.

Bridging Sync and Async with the CLI Entry Point

The main script serves as the synchronous entry point that bridges the async pipeline:

Python

def _infer_kind(path: Path) -> str:
    suf = path.suffix.lower()
    if suf == ".csv":
        return "csv"
    if suf in {".jsonl", ".ndjson"}:
        return "jsonl"
    return "csv"

The helper function infers the file format from its extension, defaulting to CSV for unknown formats. This keeps the main logic clean while providing reasonable fallback behavior.

Python

if __name__ == "__main__":
    data_path = Path(os.environ.get("DATA_PATH"))
    if not data_path or not data_path.exists():
        print("ERR DATA_PATH not set or file not found")
        raise SystemExit(2)

    kind = _infer_kind(data_path)

    results, errors = asyncio.run(run_pipeline(data_path, kind, workers=3, maxsize=3))
    for r in results:
        print(json.dumps(r, separators=(",", ":"), sort_keys=True))
    for e in errors:
        print(e)

The script reads the file path from the DATA_PATH environment variable, validates that the path exists, and exits with code 2 if the configuration is invalid. After inferring the format, we invoke the async pipeline using asyncio.run(), which creates a new event loop, runs the coroutine to completion, and cleans up the loop automatically.

The call to asyncio.run(run_pipeline(...)) is where the synchronous and asynchronous worlds meet. From the perspective of the synchronous __main__ block, this is a regular blocking function call that returns results when complete. Inside asyncio.run(), however, the entire async pipeline executes with concurrent tasks, queues, and cooperative scheduling. This pattern is standard for CLI tools: a thin synchronous wrapper that launches async processing and formats output.

After the pipeline completes, we output results as compact JSON and errors as plain text. The separators=(",", ":") argument removes whitespace for smaller output, while sort_keys=True ensures consistent field ordering across runs. This output format is both human-readable and machine-parseable, making it suitable for logging or piping to downstream tools.

Running the Pipeline

When we execute the script with a sample data file containing various transaction types and deliberate errors, the async pipeline processes everything concurrently and produces this output:

text

{"entry":{"account":"sales-north","flag":"high_value","id":1,"op":"add","qty":3,"sign":1,"total":{"amount":"31.50","currency":"USD"},"unit":{"amount":"10.50","currency":"USD"}},"ok":true}
{"entry":{"account":"Returns","id":2,"op":"refund","qty":1,"sign":-1,"total":{"amount":"5.00","currency":"EUR"},"unit":{"amount":"5.00","currency":"EUR"}},"ok":true}
{"entry":{"account":"Retail North","id":3,"op":"add","qty":2,"sign":1,"total":{"amount":"6.00","currency":"USD"},"unit":{"amount":"3.00","currency":"USD"}},"ok":true}
{"entry":{"account":"Retail North","flag":"high_value","id":4,"op":"add","qty":1,"sign":1,"total":{"amount":"15.00","currency":"USD"},"unit":{"amount":"15.00","currency":"USD"}},"ok":true}
{"entry":{"account":"Returns","id":5,"op":"refund","qty":2,"sign":-1,"total":{"amount":"5.00","currency":"EUR"},"unit":{"amount":"2.50","currency":"EUR"}},"ok":true}
{"entry":{"account":"Online Sales","flag":"high_value","id":6,"op":"add","qty":1,"sign":1,"total":{"amount":"25.99","currency":"USD"},"unit":{"amount":"25.99","currency":"USD"}},"ok":true}
{"entry":{"account":"Support Team","id":7,"op":"refund","qty":1,"sign":-1,"total":{"amount":"12.50","currency":"EUR"},"unit":{"amount":"12.50","currency":"EUR"}},"ok":true}
{"entry":{"account":"Wholesale","flag":"high_value","id":8,"op":"add","qty":5,"sign":1,"total":{"amount":"500.00","currency":"USD"},"unit":{"amount":"100.00","currency":"USD"}},"ok":true}
{"entry":{"account":"Valid Account","flag":"high_value","id":13,"op":"add","qty":1,"sign":1,"total":{"amount":"10.00","currency":"XYZ"},"unit":{"amount":"10.00","currency":"XYZ"}},"ok":true}
ERR 9 amount must be > 0.00
ERR 10 op must be 'add' or 'refund'
ERR 11 account must be non-empty
ERR 12 [<class 'decimal.ConversionSyntax'>]

The output demonstrates the pipeline's end-to-end functionality. Valid transactions appear as JSON objects with the ok: true flag and complete entry details, while errors appear as plain text lines prefixed with "ERR" and the record ID. This mixed format is intentional; it makes successful processing machine-readable while keeping errors human-friendly for quick diagnosis.

Analyzing the Output

The output reveals how the async pipeline processes records while maintaining correctness. Notice that all successful transactions appear before error messages, even though errors occurred at various positions in the input file (records 9, 10, 11, and 12 between valid records 8 and 13). This ordering reflects our pipeline architecture: valid transactions flow through the queue to consumers, getting routed and appended to results, while errors are collected separately by the producer and appended to errors. We output each list sequentially after processing completes.

The high-value flags appear correctly on transactions 1, 4, 6, 8, and 13, all with unit amounts of 10.00 or greater. Transaction 1 shows a quantity of 3 with a unit price of 10.50 USD, correctly flagged because the routing logic examines the unit amount, not the total. Transaction 8 demonstrates this with a unit price of 100.00 USD and quantity 5, resulting in a substantial total of 500.00 USD, but the flag is triggered by the unit amount exceeding the threshold.

Refund transactions (2, 5, 7) all show negative signs (sign: -1), distinguishing them from additions. The pipeline correctly handles multiple currencies; EUR refunds appear alongside USD additions without currency conversion or validation conflicts. This demonstrates that the pipeline focuses on structural correctness and business logic, leaving domain-specific rules like currency consistency to downstream systems.

The error messages reveal the validation layers working together. Error 9 indicates an amount validation failure caught by the Money or Transaction dataclass validation. Error 10 shows an unsupported operation type caught during normalization. Error 11 reveals an empty account name, triggering the account validation in the Transaction constructor. Error 12 shows a decimal parsing exception, indicating the amount field contained non-numeric data that couldn't be converted.

Transaction 13, appearing after multiple errors, confirms the pipeline's resilience. Despite encountering invalid records, processing continued through the entire file, collecting valid transactions and errors independently. This behavior is crucial for ETL systems; a single bad record shouldn't halt processing of thousands of valid records.

Conclusion and Next Steps

Excellent work! You've built a complete async ETL pipeline with backpressure, structured concurrency, and graceful shutdown coordination. The architecture demonstrates production-ready patterns: bounded queues naturally regulate flow, sentinels enable clean shutdown, TaskGroups ensure proper task lifecycle management, and the separation between the sync entry point and async core logic keeps the codebase maintainable.

The patterns you've learned scale beyond this CLI tool. Bounded queues with backpressure apply to any producer-consumer system, from web scrapers to message processors. The sentinel pattern works for any scenario where workers need coordinated shutdown signals. TaskGroups simplify error handling in concurrent systems by automatically canceling related tasks when any task fails. The asyncio.run() bridge pattern enables incremental async adoption in existing codebases; you can introduce async processing in hot paths while keeping the surrounding infrastructure synchronous.

This completes the core curriculum for building async ETL pipelines. You've progressed from self-validating domain models through streaming parsers, declarative routers, and finally to concurrent async processing with backpressure. Each lesson built on the previous, adding layers of capability while maintaining clean separation of concerns. The result is a robust, testable, and extensible foundation for data processing tools.

Now it's time to put these concepts into practice! The upcoming exercises will challenge you to implement each component step by step, building the async pipeline from the ground up. You'll start by completing the consumer coroutine with proper task coordination, then implement the producer with sentinel handling, orchestrate everything with TaskGroups, and finally bridge the sync-async boundary in the CLI entry point. These hands-on exercises will solidify your understanding and give you the confidence to build production-grade async systems in your own projects!

Introduction

The Producer-Consumer Pattern with Backpressure

Join the 1M+ learners on CodeSignal

The Sentinel Pattern for Graceful Shutdown

Implementing the Consumer Coroutine

Implementing the Producer Coroutine

Orchestrating with TaskGroup

Bridging Sync and Async with the CLI Entry Point

Running the Pipeline

Analyzing the Output

Backpressure in Action

Coordination Guarantees

Conclusion and Next Steps