Deviopath › Python

Producer and Consumers with asyncio.Queue

Hard · Async

In the producer/consumer pattern a producer puts work on an asyncio.Queue, and a fixed number of worker tasks take items off it with await queue.get(). Job runners and web crawlers use it to keep a steady number of things in flight.

The starter has a fake async handle(order) that packs an order (it sleeps a little) and returns "<id>: packed <item>". For an order with an empty item it raises ValueError.

Write async def process_orders(orders, workers):

- put every order on an asyncio.Queue and run workers worker tasks that call handle

- return the results in the same order as orders

- if handle raises, the result for that order is "failed: <message>", and the worker keeps going

- before returning, stop the workers cleanly: either put one None per worker on the queue as a stop signal, or wait with queue.join(), then cancel the workers and await them

The driver check_orders(orders, workers) is already written. It returns your results, max_active (the most orders handled at the same time, which should be min(workers, len(orders))) and leftover_tasks, the number of tasks still alive after you return (it should be 0).