Queues are fundamental data structures used for managing items in a specific order, typically First-In, First-Out (FIFO). Python provides the thread-safe queue.Queue class for multi-threaded applications and the highly efficient collections.deque for single-threaded scenarios (though queue.Queue often uses a deque internally).
This guide demonstrates common operations like accessing items without removing them, getting/removing items, checking for existence, adding multiple items, iterating, and handling exceptions.
Choosing Your Queue: queue.Queue vs. collections.deque
queue.Queue: Found in thequeuemodule. Designed for thread-safe communication between producer and consumer threads. Methods likeput()andget()include locking mechanisms. It often uses adequeinternally for storage. Use this in multi-threaded applications.collections.deque: A double-ended queue optimized for fast appends and pops from both ends. It's not inherently thread-safe but generally faster for single-threaded use cases like breadth-first search or simple FIFO buffering.
This guide covers operations on both where applicable.
Access Item Without Removing (Peeking)
Sometimes you need to inspect an item (like the first or Nth) without taking it out of the queue.
Using queue.Queue.queue[index]
The queue.Queue object has an underlying container accessible via its queue attribute (usually a deque). You can index this container.
import queue
my_queue = queue.Queue()
my_queue.put('apple')
my_queue.put('banana')
my_queue.put('cherry')
try:
first_item = my_queue.queue[0]
second_item = my_queue.queue[1]
print(f"Queue content (via .queue): {list(my_queue.queue)}")
print(f"First item (peek): '{first_item}'") # Output: 'apple'
print(f"Second item (peek): '{second_item}'") # Output: 'banana'
print(f"Getting item now: '{my_queue.get()}'") # Output: 'apple'
print(f"Queue content after get: {list(my_queue.queue)}") # Output: ['banana', 'cherry']
except IndexError:
print("Cannot peek at index, queue might be shorter.")Output:
Queue content (via .queue): ['apple', 'banana', 'cherry']
First item (peek): 'apple'
Second item (peek): 'banana'
Getting item now: 'apple'
Queue content after get: ['banana', 'cherry']my_queue.queue: Accesses the internaldeque(or sometimes list) storing the items.[0],[1], etc.: Standard indexing applies to this underlying container.
Using collections.deque[index]
deque objects directly support indexing.
from collections import deque
my_deque = deque(['red', 'green', 'blue'])
try:
first_item = my_deque[0]
last_item = my_deque[-1]
print(f"Deque content: {my_deque}")
print(f"First item (peek): '{first_item}'") # Output: 'red'
print(f"Last item (peek): '{last_item}'") # Output: 'blue'
print(f"Popping left: '{my_deque.popleft()}'") # Output: 'red'
print(f"Deque content after popleft: {my_deque}") # Output: deque(['green', 'blue'])
except IndexError:
print("Cannot peek at index, deque might be empty or index out of range.")Output:
Deque content: deque(['red', 'green', 'blue'])
First item (peek): 'red'
Last item (peek): 'blue'
Popping left: 'red'
Deque content after popleft: deque(['green', 'blue'])Get and Remove the First Item
This is the standard queue "dequeue" operation.
Using queue.Queue.get()
Removes and returns the item from the "front" of the queue (FIFO). Blocks by default if the queue is empty.
import queue
my_queue = queue.Queue()
my_queue.put('task1')
my_queue.put('task2')
print(f"Queue before get(): {list(my_queue.queue)}")
item1 = my_queue.get()
print(f"Got item: '{item1}'") # Output: 'task1'
print(f"Queue after first get(): {list(my_queue.queue)}") # Output: ['task2']
item2 = my_queue.get()
print(f"Got item: '{item2}'") # Output: 'task2'
print(f"Queue after second get(): {list(my_queue.queue)}") # Output: []Output:
Queue before get(): ['task1', 'task2']
Got item: 'task1'
Queue after first get(): ['task2']
Got item: 'task2'
Queue after second get(): []Using collections.deque.popleft()
Removes and returns the item from the left end of the deque. Raises IndexError if the deque is empty.
from collections import deque
my_deque = deque(['job1', 'job2', 'job3'])
print(f"Deque before popleft(): {my_deque}")
item1 = my_deque.popleft()
print(f"Popped left: '{item1}'") # Output: 'job1'
print(f"Deque after popleft(): {my_deque}") # Output: deque(['job2', 'job3'])
item_right = my_deque.pop()
print(f"Popped right: '{item_right}'") # Output: 'job3'
print(f"Deque after pop(): {my_deque}") # Output: deque(['job2'])Output:
Deque before popleft(): deque(['job1', 'job2', 'job3'])
Popped left: 'job1'
Deque after popleft(): deque(['job2', 'job3'])
Popped right: 'job3'
Deque after pop(): deque(['job2'])Check if an Element Exists in the Queue (in)
Using value in queue.Queue.queue
You need to check against the underlying queue.queue deque object.
import queue
my_queue = queue.Queue()
my_queue.put('apple')
my_queue.put('banana')
search_term = 'banana'
if search_term in my_queue.queue:
print(f"'{search_term}' FOUND in the queue.") # This is executed
else:
print(f"'{search_term}' NOT found in the queue.")
search_term_absent = 'kiwi'
if search_term_absent not in my_queue.queue:
print(f"'{search_term_absent}' correctly NOT found in the queue.") # This is executedOutput:
'banana' FOUND in the queue.
'kiwi' correctly NOT found in the queue.Using value in collections.deque
Deques directly support the in operator.
from collections import deque
my_deque = deque(['red', 'green', 'blue'])
search_term = 'green'
if search_term in my_deque:
print(f"'{search_term}' FOUND in the deque.") # This is executed
else:
print(f"'{search_term}' NOT found in the deque.")Output:
'green' FOUND in the deque.Add Multiple Items to the Queue
Using a Loop with queue.Queue.put()
The queue.Queue doesn't have a multi-item add method, so you use a loop.
import queue
my_queue = queue.Queue()
items_to_add = ['data1', 'data2', 'data3']
print(f"Queue before adding multiple: {list(my_queue.queue)}")
for item in items_to_add:
my_queue.put(item)
print(f"Queue after adding multiple: {list(my_queue.queue)}")Output:
Queue before adding multiple: []
Queue after adding multiple: ['data1', 'data2', 'data3']Using collections.deque.extend()
Deques have an efficient extend() method to add all items from an iterable to the right side.
from collections import deque
my_deque = deque(['initial'])
items_to_add = ['new1', 'new2', 'new3']
print(f"Deque before extend: {my_deque}")
my_deque.extend(items_to_add)
print(f"Deque after extend: {my_deque}")Output:
Deque before extend: deque(['initial'])
Deque after extend: deque(['initial', 'new1', 'new2', 'new3'])Iterate Through a Queue
Iterating and Consuming (while not q.empty(): q.get())
This pattern processes items one by one, removing them from the queue until it's empty. Suitable for worker patterns.
import queue
my_queue = queue.Queue()
for i in range(3): my_queue.put(f'item_{i}')
print("Iterating and consuming queue:")
while not my_queue.empty():
item = my_queue.get()
print(f" Processing item: {item}")
print(f"Queue after consuming loop is empty: {my_queue.empty()}") # Output: TrueOutput:
Iterating and consuming queue:
Processing item: item_0
Processing item: item_1
Processing item: item_2
Queue after consuming loop is empty: TrueIterating Without Consuming (for item in q.queue)
To loop through the items without removing them, iterate over the underlying queue.queue deque.
import queue
my_queue = queue.Queue()
for i in range(3): my_queue.put(f'element_{i}')
print("Iterating without consuming (via .queue):")
for item in my_queue.queue:
print(f" Observing item: {item}")
print(f"Queue after non-consuming loop: {list(my_queue.queue)}")Output:
Iterating without consuming (via .queue):
Observing item: element_0
Observing item: element_1
Observing item: element_2
Queue after non-consuming loop: ['element_0', 'element_1', 'element_2']Handle queue.Empty Exception
When using queue.Queue.get(block=False) or queue.Queue.get_nowait(), these methods will raise a queue.Empty exception if the queue has no items available immediately. Use try...except to handle this.
import queue
my_queue = queue.Queue() # Start with an empty queue
print("Attempting non-blocking get:")
try:
item = my_queue.get(block=False)
print(f"Got item: {item}")
except queue.Empty:
print("Caught queue.Empty: The queue is currently empty.")
except Exception as e:
print(f"Caught unexpected error: {e}")
try:
item = my_queue.get(block=False)
except queue.Empty:
print("queue.Empty exception in second try")
else:
print(f"Successfully got item: {item} (in else block)")Output:
Attempting non-blocking get:
Caught queue.Empty: The queue is currently empty.
queue.Empty exception in second tryConclusion
Working with queues in Python involves choosing between the thread-safe queue.Queue and the efficient collections.deque.
- Peeking (access without removal): Use indexing on
q.queueormy_deque. - Getting/Removing: Use
q.get()(blocking, thread-safe) ormy_deque.popleft()(non-blocking, faster, raisesIndexErrorif empty). - Checking Existence: Use
value in q.queueorvalue in my_deque. - Adding Multiple: Loop with
q.put()or usemy_deque.extend(). - Iterating: Use
while not q.empty(): item = q.get()to consume, orfor item in q.queue:to observe without consuming. - Handling Emptiness: Catch
queue.Emptywhen using non-blockingget()methods.
Understanding these operations allows effective use of queues for managing ordered data in various Python applications.