Python Basics: Managing Pending Work with queue

A background worker needs a safe way to receive jobs and tell its caller when processing is finished. Python’s queue module provides synchronized queues, including a bounded FIFO queue and a priority queue.

These examples separate three questions: whether another job fits, whether submitted work has been acknowledged, and which pending job should be taken next. They use only local values and print results after processing rather than interleaving output from multiple threads.

Prepare a scratch directory

These three standalone examples were tested on Linux with Python 3.12.14. The examples require Python 3.6 or newer because the worker formats results with f-strings. They use long-established Queue methods, not the shutdown API added in Python 3.13. Use Bash, a text editor, and the standard mktemp, rm, and rmdir utilities. No third-party packages are required. Keep assertions enabled by running without Python’s -O option.

Create a fresh directory and keep using the same terminal:

start_dir="$PWD"
work_dir="$(mktemp -d)" || exit 1
cd -- "$work_dir" || exit 1

Save each named source file here as UTF-8.

Avoid source filenames such as queue.py or threading.py. The worker handles invented inventory counts; it does not change a real inventory or write files.

1. Handle full and empty queues without blocking

Save as bound_pending_jobs.py:

from queue import Empty, Full, Queue


jobs = Queue(maxsize=2)
jobs.put_nowait("resize")
jobs.put_nowait("thumbnail")
try:
    jobs.put_nowait("archive")
except Full:
    print("Extra job: queue full")
else:
    raise AssertionError("A third job fit in a two-slot queue")

received = []
for _ in range(2):
    received.append(jobs.get_nowait())
    jobs.task_done()
jobs.join()
assert received == ["resize", "thumbnail"]
print("Received:", received)
try:
    jobs.get_nowait()
except Empty:
    print("Next job: queue empty")
else:
    raise AssertionError("An empty queue returned a job")

Run it:

python3 bound_pending_jobs.py

Output:

Extra job: queue full
Received: ['resize', 'thumbnail']
Next job: queue empty

The two-slot queue accepts resize and thumbnail, then immediately raises Full for archive. get_nowait retrieves the first two jobs in FIFO order and raises Empty after they are gone. No check-then-act race is needed.

Do not use empty, full, or qsize as a promise that a later operation cannot block when other threads are involved. Attempt the operation and handle its result. Each retrieved job is acknowledged exactly once; join then has no unfinished work to wait for.

2. Finish a worker and acknowledge every job

Save as finish_worker_queue.py:

from queue import Queue
from threading import Thread


jobs = Queue(maxsize=2)
results = []
stop = object()


def worker():
    while True:
        item = jobs.get()
        try:
            if item is stop:
                return
            name, quantity_text = item
            try:
                quantity = int(quantity_text)
                if quantity < 0:
                    raise ValueError("negative quantity")
            except ValueError:
                results.append(f"{name}: rejected")
            else:
                results.append(f"{name}: {quantity} units")
        finally:
            jobs.task_done()


thread = Thread(target=worker)
thread.start()
for job in [("boxes", "4"), ("labels", "bad"), ("bags", "0")]:
    jobs.put(job)
jobs.put(stop)
jobs.join()
thread.join()
assert results == ["boxes: 4 units", "labels: rejected", "bags: 0 units"]
assert not thread.is_alive()
print("\n".join(results))
print("Worker stopped:", not thread.is_alive())

Run it:

python3 finish_worker_queue.py

Output:

boxes: 4 units
labels: rejected
bags: 0 units
Worker stopped: True

The worker starts before the producer fills the bounded queue. A unique sentinel object ends its loop after the submitted jobs. The finally block acknowledges every retrieved item, including the rejected count and the sentinel, so no accounting entry is left unfinished.

Queue.join waits for acknowledgements; Thread.join waits for the thread itself to end. The results list has one writer and is read only after both waits. For several workers, send one sentinel per worker and do not assume completion order will match submission order. Unexpected worker crashes also need a supervisor in a real service.

3. Break equal-priority ties without comparing payloads

Save as order_priority_jobs.py:

from itertools import count
from queue import PriorityQueue


jobs = PriorityQueue()
sequence = count()
for priority, job in [
    (5, {"name": "weekly report"}),
    (1, {"name": "failed upload"}),
    (1, {"name": "missing preview"}),
]:
    jobs.put((priority, next(sequence), job))

ordered = []
for _ in range(3):
    priority, position, job = jobs.get_nowait()
    ordered.append(job["name"])
    jobs.task_done()
jobs.join()
assert ordered == ["failed upload", "missing preview", "weekly report"]
print("Processing order:", ordered)
print("Equal priorities retain submission order:", ordered[:2])

Run it:

python3 order_priority_jobs.py

Output:

Processing order: ['failed upload', 'missing preview', 'weekly report']
Equal priorities retain submission order: ['failed upload', 'missing preview']

Lower priority numbers come out first. The sequence number creates a unique second comparison field, preserving submission order among equal-priority jobs. The dictionaries are payloads and never need to be compared to one another.

Without the sequence field, equal priorities could make Python try to order dictionaries and raise TypeError. The fixed three-item loop is appropriate for this closed demonstration. An ongoing producer-consumer system needs an explicit stopping protocol rather than guessing how many jobs will arrive.

A queue coordinates access to its entries; it does not automatically protect mutable objects after a consumer receives them. Decide which worker owns each payload, distinguish rejected jobs from successful ones, and keep task_done paired with successful get calls.

Clean up

In the same terminal, remove only these source files, return to the original directory, and remove the now-empty scratch directory:

rm -- bound_pending_jobs.py finish_worker_queue.py order_priority_jobs.py
cd -- "$start_dir" || exit 1
rmdir -- "$work_dir"
unset start_dir work_dir

If rmdir reports that anything remains, inspect the directory rather than deleting unknown files.

References

• Python 3.12 queue documentation
• Queue task accounting
• Thread lifecycle and join

Acknowledgements

This post was created with assistance from AI.