How to Create a Queue Using collections.deque in Python
Queues are fundamental data structures that process elements in First-In, First-Out (FIFO) order, meaning the first item added is the first one removed. From task schedulers to message processing systems, queues appear throughout software development. While Python lists can technically function as queues, they suffer from significant performance issues at scale.
This guide demonstrates how to implement efficient queues using Python's collections.deque, explains why it outperforms lists for this purpose, and shows practical real-world examples including a reusable queue class and a task processor.
Why deque Outperforms list for Queues
The critical difference lies in how each structure handles removal from the front. When you call list.pop(0), Python must shift every remaining element one position to the left. This is an O(n) operation that becomes increasingly slow as the list grows.
The collections.deque (double-ended queue) is specifically optimized for operations at both ends. Its popleft() method runs in O(1) constant time, maintaining consistent performance whether your queue contains ten items or ten million.
Demonstrating the Problem with Lists
queue = ["first", "second", "third"]
# Removing from the front requires shifting all remaining elements
item = queue.pop(0)
print(item)
print(queue)
Output:
first
['second', 'third']
While the output is correct, every element after the removed one had to be copied to a new position internally. For a list with a million items, that means roughly a million copy operations for every single dequeue.
The deque Solution
from collections import deque
queue = deque()
# Add items to the back (enqueue)
queue.append("first")
queue.append("second")
queue.append("third")
# Remove items from the front (dequeue)
print(queue.popleft())
print(queue.popleft())
print(queue)
Output:
first
second
deque(['third'])
The popleft() method removes the front element in constant time regardless of the queue size, making deque the correct choice for any queue implementation.
Basic Queue Operations with deque
Here are the essential operations you need for a FIFO queue:
from collections import deque
queue = deque()
# Enqueue: add items to the back
queue.append("task_1")
queue.append("task_2")
queue.append("task_3")
print(f"Queue after enqueuing: {queue}")
# Peek: view the front item without removing it
front = queue[0]
print(f"Front item: {front}")
# Dequeue: remove and return the front item
processed = queue.popleft()
print(f"Processed: {processed}")
# Check size
print(f"Remaining items: {len(queue)}")
# Check if empty
print(f"Is empty: {len(queue) == 0}")
Output:
Queue after enqueuing: deque(['task_1', 'task_2', 'task_3'])
Front item: task_1
Processed: task_1
Remaining items: 2
Is empty: False
Building a Reusable Queue Class
Wrapping deque in a custom class provides intuitive method names and encapsulates the implementation details:
from collections import deque
class Queue:
"""A FIFO queue implementation using deque."""
def __init__(self):
self._items = deque()
def enqueue(self, item):
"""Add an item to the back of the queue."""
self._items.append(item)
def dequeue(self):
"""Remove and return the front item, or None if empty."""
if not self._items:
return None
return self._items.popleft()
def peek(self):
"""View the front item without removing it."""
return self._items[0] if self._items else None
def is_empty(self):
"""Check if the queue has no items."""
return len(self._items) == 0
def __len__(self):
return len(self._items)
def __repr__(self):
return f"Queue({list(self._items)})"
# Usage
task_queue = Queue()
task_queue.enqueue("Send welcome email")
task_queue.enqueue("Process payment")
task_queue.enqueue("Generate invoice")
print(f"Next task: {task_queue.peek()}")
print(f"Processing: {task_queue.dequeue()}")
print(f"Remaining: {len(task_queue)}")
print(task_queue)
Output:
Next task: Send welcome email
Processing: Send welcome email
Remaining: 2
Queue(['Process payment', 'Generate invoice'])
The underscore prefix in self._items signals to other developers that this is an internal implementation detail. Users of the class should interact through the public methods (enqueue, dequeue, peek) rather than accessing the underlying deque directly.
A Common Mistake: Dequeuing from an Empty Queue
If you call popleft() on an empty deque without checking first, Python raises an IndexError:
from collections import deque
queue = deque()
item = queue.popleft()
Output:
IndexError: pop from an empty deque
The custom Queue class above handles this by returning None when the queue is empty. Alternatively, you can raise a custom exception for stricter error handling:
def dequeue(self):
if not self._items:
raise IndexError("Cannot dequeue from an empty queue")
return self._items.popleft()
Using maxlen for Bounded Queues
The deque constructor accepts an optional maxlen parameter that limits the queue size. When the queue is full and a new item is added, the oldest item is automatically discarded:
from collections import deque
# Keep only the 3 most recent log entries
recent_logs = deque(maxlen=3)
recent_logs.append("Log entry 1")
recent_logs.append("Log entry 2")
recent_logs.append("Log entry 3")
print(recent_logs)
# Adding a fourth entry automatically removes the first
recent_logs.append("Log entry 4")
print(recent_logs)
Output:
deque(['Log entry 1', 'Log entry 2', 'Log entry 3'], maxlen=3)
deque(['Log entry 2', 'Log entry 3', 'Log entry 4'], maxlen=3)
This is useful for scenarios like keeping a sliding window of recent events, caching the last N results, or implementing a fixed-size buffer.
Performance Comparison
| Operation | list | deque | Winner |
|---|---|---|---|
| Add to back | .append() O(1) | .append() O(1) | Tie |
| Remove from front | .pop(0) O(n) | .popleft() O(1) | deque |
| Add to front | .insert(0, x) O(n) | .appendleft() O(1) | deque |
| Access by index | O(1) | O(n) | list |
The one area where list outperforms deque is random index access. If you frequently need to access elements by position (e.g., queue[5]), a list may be more appropriate. For pure queue operations (add to back, remove from front), deque is always the better choice.
Practical Example: Task Processor
Here is a more realistic example that simulates a task processing system with status tracking:
from collections import deque
import time
class TaskProcessor:
"""Process tasks in FIFO order with status tracking."""
def __init__(self):
self._pending = deque()
self._completed = []
def add_task(self, task_name):
"""Queue a new task for processing."""
self._pending.append({
"name": task_name,
"queued_at": time.time()
})
print(f"Queued: {task_name}")
def process_next(self):
"""Process the next task in line."""
if not self._pending:
print("No tasks pending")
return None
task = self._pending.popleft()
task["completed_at"] = time.time()
self._completed.append(task)
print(f"Completed: {task['name']}")
return task
@property
def pending_count(self):
return len(self._pending)
@property
def completed_count(self):
return len(self._completed)
# Demonstration
processor = TaskProcessor()
processor.add_task("Validate user input")
processor.add_task("Save to database")
processor.add_task("Send confirmation")
print(f"\nTasks pending: {processor.pending_count}")
processor.process_next()
processor.process_next()
print(f"\nTasks pending: {processor.pending_count}")
print(f"Tasks completed: {processor.completed_count}")
Output:
Queued: Validate user input
Queued: Save to database
Queued: Send confirmation
Tasks pending: 3
Completed: Validate user input
Completed: Save to database
Tasks pending: 1
Tasks completed: 2
Choosing the Right Queue Implementation
Python offers several queue implementations for different scenarios:
| Use Case | Recommended Tool |
|---|---|
| General-purpose FIFO queue | collections.deque |
| Multi-threaded producer/consumer | queue.Queue |
| Priority-based processing | heapq or queue.PriorityQueue |
| Async applications | asyncio.Queue |
While individual deque operations like append() and popleft() are atomic in CPython due to the GIL, complex operations involving multiple method calls are not thread-safe. For producer-consumer patterns with multiple threads, use queue.Queue, which provides built-in locking mechanisms:
from queue import Queue
thread_safe_queue = Queue()
thread_safe_queue.put("item")
item = thread_safe_queue.get()
Conclusion
The collections.deque is the right foundation for queue implementations in Python.
- It provides O(1) constant-time performance for both
append()andpopleft(), whereaslist.pop(0)degrades to O(n) as the queue grows. - Wrap it in a custom class for clean, reusable queue abstractions, use
maxlenfor bounded queues that automatically discard old items, and switch toqueue.Queuewhen thread safety is required. - By choosing
dequeoverlistfor queue operations, you ensure your applications remain fast and scalable regardless of data volume.