How to Dump a Queue to a List in Python

Python provides different queue implementations for different use cases. The collections.deque is a general-purpose double-ended queue, while queue.Queue and its variants are designed for thread-safe communication between threads. Converting these queues to lists requires understanding their internal structure and thread-safety considerations.

In this guide, you will learn how to convert each type of queue to a list, choose between destructive and non-destructive approaches, handle thread safety correctly, and work with priority queues.

Converting collections.deque

The deque is Python's general-purpose queue and the most commonly used. It is directly iterable, making conversion to a list straightforward:

from collections import deque

q = deque([1, 2, 3, 4, 5])

# Direct conversion - queue remains intact
my_list = list(q)

print(my_list)
print(q)

Output:

[1, 2, 3, 4, 5]
deque([1, 2, 3, 4, 5])

The original deque is unchanged after the conversion. You can also use unpacking syntax for the same result:

from collections import deque

q = deque([1, 2, 3, 4, 5])

my_list = [*q]

print(my_list)

Output:

[1, 2, 3, 4, 5]

Both approaches create a new list containing copies of the references in the deque.

Converting queue.Queue

The queue.Queue class is designed for thread-safe producer-consumer patterns. Unlike deque, it is not directly iterable. You can access its internal buffer through the .queue attribute:

from queue import Queue

q = Queue()
q.put("first")
q.put("second")
q.put("third")

# Access the internal deque buffer
my_list = list(q.queue)

print(my_list)

Output:

['first', 'second', 'third']

warning

Accessing q.queue directly bypasses the thread-safety mechanisms that Queue provides. In multi-threaded code, this can cause race conditions if another thread modifies the queue while you are reading it. Only use this approach when you are certain no other threads are modifying the queue simultaneously.

Thread-Safe Conversion by Draining

For truly thread-safe conversion in concurrent applications, consume the queue properly using get_nowait():

from queue import Queue, Empty

def drain_queue(q):
    """Safely extract all items from a thread-safe queue."""
    items = []
    while True:
        try:
            item = q.get_nowait()
            items.append(item)
            q.task_done()
        except Empty:
            break
    return items

q = Queue()
q.put("A")
q.put("B")
q.put("C")

result = drain_queue(q)
print(result)
print(f"Queue empty: {q.empty()}")

Output:

['A', 'B', 'C']
Queue empty: True

This approach uses the thread-safe get_nowait() method for each item, ensuring no race conditions. However, it empties the queue in the process.

Destructive vs Non-Destructive Conversion

Choosing between these two approaches depends on whether you need to preserve the queue contents after conversion.

Non-Destructive (Queue Keeps Its Items)

from collections import deque

q = deque([1, 2, 3])

preserved_list = list(q)

print(f"List: {preserved_list}")
print(f"Queue length: {len(q)}")

Output:

List: [1, 2, 3]
Queue length: 3

Destructive (Queue Is Emptied)

from collections import deque

q = deque([1, 2, 3])

consumed_list = []
while q:
    consumed_list.append(q.popleft())

print(f"List: {consumed_list}")
print(f"Queue length: {len(q)}")

Output:

List: [1, 2, 3]
Queue length: 0

For queue.Queue objects, the destructive approach uses get():

from queue import Queue

q = Queue()
for i in range(3):
    q.put(i)

result = []
while not q.empty():
    result.append(q.get())

print(f"List: {result}")
print(f"Queue empty: {q.empty()}")

Output:

List: [0, 1, 2]
Queue empty: True

Priority Queue Conversion

PriorityQueue stores items in a heap structure. Accessing the internal buffer with list(pq.queue) gives you heap order, which is not necessarily fully sorted:

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((3, "low"))
pq.put((1, "high"))
pq.put((2, "medium"))

# Internal heap order (may not be fully sorted)
raw_list = list(pq.queue)
print(f"Heap order: {raw_list}")

# To get guaranteed sorted order, drain the queue
sorted_items = []
while not pq.empty():
    sorted_items.append(pq.get())

print(f"Sorted order: {sorted_items}")

Output:

Heap order: [(1, 'high'), (3, 'low'), (2, 'medium')]
Sorted order: [(1, 'high'), (2, 'medium'), (3, 'low')]

Notice that the heap order has (3, 'low') before (2, 'medium'), which is valid for a heap but not a fully sorted sequence.

tip

If you need items in correct priority order, drain the PriorityQueue with get() rather than reading the internal buffer directly. The get() method always returns the highest-priority (lowest value) item.

Practical Example: Batch Processing

A common real-world pattern is collecting queued items into batches for more efficient processing:

from queue import Queue, Empty

def process_batch(items):
    print(f"Processing batch of {len(items)} items: {items}")

def batch_processor(q, batch_size=5, timeout=0.5):
    """Collect items into batches for processing."""
    batch = []

    while True:
        try:
            item = q.get(timeout=timeout)
            batch.append(item)
            q.task_done()

            if len(batch) >= batch_size:
                process_batch(batch)
                batch = []

        except Empty:
            # Timeout reached - process any remaining items
            if batch:
                process_batch(batch)
            break

q = Queue()
for i in range(12):
    q.put(f"item_{i}")

batch_processor(q)

Output:

Processing batch of 5 items: ['item_0', 'item_1', 'item_2', 'item_3', 'item_4']
Processing batch of 5 items: ['item_5', 'item_6', 'item_7', 'item_8', 'item_9']
Processing batch of 2 items: ['item_10', 'item_11']

The function collects items until the batch is full, processes them, and handles any remaining items when the queue is empty.

A Common Mistake: Using q.empty() in Multi-Threaded Code

A frequent error in concurrent programs is using q.empty() as a loop condition:

from queue import Queue

q = Queue()

# Dangerous in multi-threaded code!
while not q.empty():
    item = q.get()  # Another thread might have taken the item between the check and get

Between the q.empty() check and the q.get() call, another thread could remove the last item, causing get() to block indefinitely. Use get_nowait() with exception handling instead:

from queue import Queue, Empty

q = Queue()

try:
    item = q.get_nowait()
except Empty:
    print("Queue is empty")

In single-threaded code, while not q.empty() is perfectly safe.

Queue Type Comparison

Queue Type	Non-Destructive Conversion	Thread Safe	Typical Use Case
deque	list(q)	No	General purpose, single-threaded
Queue	list(q.queue) (unsafe)	Yes	Producer-consumer patterns
LifoQueue	list(q.queue) (unsafe)	Yes	Stack behavior (LIFO)
PriorityQueue	list(q.queue) (heap order)	Yes	Priority-based processing

Conclusion

• Use list(q) for deque objects. It is simple, efficient, and non-destructive.
• For queue.Queue and its variants, access the internal buffer with list(q.queue) for a quick non-destructive snapshot, but only when no other threads are modifying the queue.
• In concurrent code, drain the queue using get_nowait() in a loop to ensure thread-safe access.
• For PriorityQueue, always drain with get() if you need items in correct priority order, since the internal heap structure does not guarantee full sorting.