Joinable PriorityQueue in python's asyncio

Question

As per the documentation, there are several implementations of Queues. The ones I'm interested in are the JoinableQueue and the PriorityQueue, as I'd like to have a joinable queue with priorities.

It appears that I can only get one of the functionalities in the lower versions. I.e. in 3.5 I can make the distinciton with Queue (which is joinable) and PriorityQueue, but in python below 3.5 there is JoinableQueue and PriorityQueue (see example below).

Is it possible to combine them to get a general way to get a joinable PriorityQueue in 3.4?

try:
    # Python 3.4.
    from asyncio import JoinableQueue as Queue # joinable
    from asyncio import PriorityQueue          # I assume this one is not joinable
except ImportError:
    # Python 3.5.
    from asyncio import Queue                  # standard joinable
    from asyncio import PriorityQueue          # I assume this is the one I want

Another approach would be to somehow perhaps influence Queue?

Answer 1

Because of the way JoinableQueue and PriorityQueue are implemented, you can get a JoinablePriorityQueue by subclassing both via multiple inheritance, as long as you list JoinableQueue first.

The reason this works is that PriorityQueue is very simply implemented:

class PriorityQueue(Queue):
    """A subclass of Queue; retrieves entries in priority order (lowest first).

    Entries are typically tuples of the form: (priority number, data).
    """

    def _init(self, maxsize):
        self._queue = []

    def _put(self, item, heappush=heapq.heappush):
        heappush(self._queue, item)

    def _get(self, heappop=heapq.heappop):
        return heappop(self._queue)

While JoinableQueue is more complicated, the only method both it and PriorityQueue implement is _put, and crucially, JoinableQUeue calls super()._put(..) in its own put implementation, which means it will cooperate with PriorityQueue properly.

Here's an example demonstrating that it works:

from asyncio import PriorityQueue, JoinableQueue
import asyncio
import random

class JoinablePriorityQueue(JoinableQueue, PriorityQueue):
    pass


@asyncio.coroutine
def consume(q):
    while True:
        a = yield from q.get()
        print("got a {}".format(a))
        if a[1] is None:
            q.task_done()
            return
        asyncio.sleep(1)
        q.task_done()

@asyncio.coroutine
def produce(q):
    for i in range(10):
        yield from q.put((random.randint(0,10), i))
    yield from q.put((100, None)) # Will be last
    asyncio.async(consume(q))
    print("waiting...")
    yield from q.join()
    print("waited")

loop = asyncio.get_event_loop()
q = JoinablePriorityQueue()
loop.run_until_complete(produce(q))

Output:

waiting...
got a (1, 2)
got a (2, 1)
got a (4, 4)
got a (5, 0)
got a (6, 8)
got a (6, 9)
got a (8, 3)
got a (9, 5)
got a (9, 7)
got a (10, 6)
got a (100, None)
waited