Producer-consumer model in python with asyncio

async_test.py

import asyncio
import random
import time


# emulate a time consuming task
async def task_A(char):
    time_to_sleep = random.randint(1, 3)
    id = f'Product_{char}[{time.strftime("%M:%S", time.localtime())}, {time_to_sleep}s]'
    print("Producing: " + id)
    await asyncio.to_thread(
        time.sleep, time_to_sleep
    )  # a non-async task that takes 1-3 seconds
    return id


# play the result of task_A
async def play_result_A(result_A):
    print(
        f"At {time.strftime('%M:%S', time.localtime())}, Start Playing: {result_A} for 4s"
    )
    await asyncio.to_thread(time.sleep, 4)  # takes 4 seconds to play
    print(f"At {time.strftime('%M:%S', time.localtime())} Finish_Playing: {result_A}")


async def main(chat_completions):
    async def workerA(queue):
        for char in chat_completions:
            result = await task_A(char)
            await queue.put(result)

    async def player(queue):
        while True:
            result = await queue.get()
            await play_result_A(result)
            queue.task_done()

    queue = asyncio.Queue()
    producer = asyncio.create_task(workerA(queue))
    consumer = asyncio.create_task(player(queue))

    await producer
    await queue.join()  # wait until all tasks are done
    consumer.cancel()


# usage
if __name__ == "__main__":
    chat_completions = iter("ABCDEFGHIJ")  # emulate chat_completions iterator
    asyncio.run(main(chat_completions))

Console output:

Producing: Product_A[18:08, 3s]
Producing: Product_B[18:11, 2s]
At 18:11, Start Playing: Product_A[18:08, 3s] for 4s
Producing: Product_C[18:13, 2s]
At 18:15 Finish_Playing: Product_A[18:08, 3s]
At 18:15, Start Playing: Product_B[18:11, 2s] for 4s
Producing: Product_D[18:15, 1s]
Producing: Product_E[18:16, 3s]
At 18:19 Finish_Playing: Product_B[18:11, 2s]
At 18:19, Start Playing: Product_C[18:13, 2s] for 4s
Producing: Product_F[18:19, 2s]
Producing: Product_G[18:21, 2s]
At 18:23 Finish_Playing: Product_C[18:13, 2s]
At 18:23, Start Playing: Product_D[18:15, 1s] for 4s
Producing: Product_H[18:23, 3s]
Producing: Product_I[18:26, 2s]
At 18:27 Finish_Playing: Product_D[18:15, 1s]
At 18:27, Start Playing: Product_E[18:16, 3s] for 4s
Producing: Product_J[18:28, 3s]
At 18:31 Finish_Playing: Product_E[18:16, 3s]
At 18:31, Start Playing: Product_F[18:19, 2s] for 4s
At 18:35 Finish_Playing: Product_F[18:19, 2s]
At 18:35, Start Playing: Product_G[18:21, 2s] for 4s
At 18:39 Finish_Playing: Product_G[18:21, 2s]
At 18:39, Start Playing: Product_H[18:23, 3s] for 4s
At 18:43 Finish_Playing: Product_H[18:23, 3s]
At 18:43, Start Playing: Product_I[18:26, 2s] for 4s
At 18:47 Finish_Playing: Product_I[18:26, 2s]
At 18:47, Start Playing: Product_J[18:28, 3s] for 4s
At 18:51 Finish_Playing: Product_J[18:28, 3s]