How to Implement a Priority Queue for Messages in Python

Build a message priority queue with heapq and dataclasses that pops messages by priority, using sequence numbers to keep insertion order.

Easy Python 3.7+ Aug 9, 2026 Streaming & messaging 15 views 0 copies

priority-queue heapq dataclass messaging streaming

Python code

32 lines

Python 3.7+

import heapq
from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class Message:
    priority: int
    sequence: int = field(compare=False)
    content: str = field(compare=False)

class PriorityQueue:
    def __init__(self):
        self._heap = []

    def push(self, priority: int, sequence: int, content: str) -> None:
        heapq.heappush(self._heap, Message(priority, sequence, content))

    def pop(self) -> Message:
        return heapq.heappop(self._heap)

    def __len__(self) -> int:
        return len(self._heap)

if __name__ == "__main__":
    q = PriorityQueue()
    q.push(3, 1, "low")
    q.push(1, 2, "high")
    q.push(2, 3, "medium")
    q.push(1, 4, "high-ish")
    while q:
        msg = q.pop()
        print(f"{msg.priority} | {msg.sequence} | {msg.content}")

Output

stdout

1 | 2 | high
1 | 4 | high-ish
2 | 3 | medium
3 | 1 | low

How it works

The @dataclass(order=True) decorator makes Message objects comparable by their fields. Since sequence and content have compare=False, comparison only uses priority, so equal priorities fall back to insertion order automatically in heapq (which is stable). The heap is a list that maintains the smallest-priority item at the front, so heappush and heappop keep it organized in O(log n) time.

Common mistakes

Forgetting `compare=False` on sequence and content causes comparisons to consider all fields, breaking FIFO order among equal priorities
Using `list.pop(0)` instead of heapq's heappop leads to O(n) time
Not including a sequence number when you need stable ordering for equal priorities

Variations

Use `PriorityQueue` from the `queue` module for thread-safe message queues
Implement with a custom comparator class instead of dataclass ordering

Real-world use cases

Prioritizing urgent events over routine jobs in a task scheduler or message broker consumer.
Processing latency-sensitive notifications before bulk analytics batches in a stream processing pipeline.
Implementing a messaging system where VIP customer messages jump the queue ahead of standard traffic.

How to Implement a Priority Queue for Messages in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes