Unleashing the Energy of Python Asyncio’s Queue | by Peng Qian | Jun, 2023

The queue talked about earlier is a First-In-First-Out (FIFO) queue, the place the primary merchandise to enter the queue is the primary to be retrieved. That is appropriate when all duties within the queue have the identical precedence.

Nonetheless, think about the next scenario:

Suppose there’s a queue with duties ready in line, every requiring a protracted processing time.

An error log or VIP person entry is a high-priority job that wants instant consideration. What ought to we do?

That is the place asyncio.PriorityQueue comes into play.

Briefly describe asyncio.PriorityQueue’s implementation

In contrast to FIFO queues primarily based on lists, asyncio.PriorityQueue is predicated on heaps. It’s constructed utilizing a binary tree construction.

It’s possible you’ll be accustomed to binary search timber, which be sure that probably the most minor node is all the time the leftmost node.

Nonetheless, the binary tree in asyncio.PriorityQueue ensures that probably the most minor node is all the time on the high, so the very best precedence node is completely eliminated first.

Actual-world instance with asyncio.PriorityQueue

Let’s illustrate the utilization of asyncio.PriorityQueue with a real-world state of affairs that exists in apply.

Think about we’ve an order service API. The API takes time for every order to course of, however we are able to’t preserve customers ready too lengthy.

So when a person locations an order, the API first places the order right into a queue, permitting a background job to course of it asynchronously whereas instantly returning a message to the person.

This API accepts orders from two varieties of customers: common customers and VIP customers. It should be sure that VIP person orders are processed with the very best precedence.

To maintain the training curve low for readers, on this instance, we’ll use aiohttp to implement the server. The precise code is as follows:

First, we outline an enumeration marking the 2 classes: common customers and VIP customers.

Subsequent, we use dataclass to outline a person’s order, which incorporates the person sort and order processing length. The order length shouldn’t be thought-about in precedence sorting.

Then we outline the patron technique process_order_worker, which retrieves orders from the queue and simulates the order processing.

Don’t overlook to make use of queue.task_done() to inform the queue that we completed processing the order.

Following that, we implement the order API utilizing aiohttp. This API responds to person requests, generates an order object, and locations it within the asyncio.PriorityQueue.

It then instantly returns a response to the person, avoiding person wait time.

When this system begins, we use create_order_queue to initialize the queue and order consumption duties.

When this system ends, we use destroy_order_queue to make sure that all orders within the queue are processed and the background duties are closed appropriately.

queue.be part of() will look forward to all the info within the queue to be processed. asyncio.wait_for units a timeout of 20 seconds, after which it can not wait queue.be part of() to finish.

We are able to take a look at this implementation utilizing PyCharm’s HTTP Request:

As you possibly can see, the 2 high-priority duties are processed as anticipated. Good!