A disk-assisted, thread-safe queue for Python — overflow to disk when memory is tight, never lose a message when the process dies.
queue.Queue lives in memory — once it's gone, your messages are gone. pydisq keeps the same friendly API but transparently pages cold items to disk, so you can:
- 📦 Buffer more than RAM allows — only
cache_sizeitems stay in memory at a time. - 💾 Survive crashes — every page is
fsync'd, and the queue recovers from its last checkpoint on restart. - ⚡ Stay fast — msgpack binary serialization keeps the disk hop cheap.
- 🧵 Work across threads — built on
threading.Lock+Condition, exactly like CPython's stdlibQueue.
from DiskQueue import DiskQueue
q = DiskQueue(path='./', queue_name='jobs', cache_size=10)
for i in range(50):
q.put({'job_id': i})
while len(q):
print(q.get())That's it. No daemon to run, no broker to install — it's just a folder on disk and a Python class.
| 🧠 Memory-first, disk-second | Items live in an in-memory ring until the cache fills, then a whole page flushes to disk. |
| 🪪 Drop-in stdlib API | put(), get(), put_nowait(), get_nowait() — same shape as queue.Queue. |
👀 peek() built in |
Inspect the next n items without consuming them. |
| 🧵 Multi-producer / multi-consumer | Condition variables coordinate blocking producers and consumers safely. |
| 🛟 Crash-safe checkpoints | Head/tail pointers live in an index file, flushed with fsync. |
| 🪶 Zero infrastructure | No Redis, no RabbitMQ, no SQLite — just a directory you choose. |
| 📦 msgpack on the wire | Smaller and faster than pickle/json for typical payloads. |
┌──────────────────┐ ┌──────────────────┐
│ put_buffer │ │ get_buffer │
│ (in memory, size │ ── flush page ──▶ │ (in memory, size │
│ = cache_size) │ │ = cache_size) │
└────────┬─────────┘ └────────▲─────────┘
│ │
▼ │
┌────────────────────────────────────────────────┴─────────┐
│ ./queue_name/ <tail> ... <head+1> <head> │
│ │ │ │ │
│ └─ index ──────┴── one msgpack file per page ───────┘
└──────────────────────────────────────────────────────────┘
Producers append to the put buffer; once it fills, the buffer is serialized to a numbered page file and tail advances. Consumers pull from the get buffer; when it empties, the next page is loaded from disk and head advances. The 000 index file records head,tail so the queue picks up exactly where it left off after a crash.
from DiskQueue import DiskQueue
import threading, time, random
q = DiskQueue(path='./', queue_name='es-miss', cache_size=4)
def producer(pid):
while True:
item = random.randint(1, 50)
print(f'[🤖 producer {pid}] put {item}')
q.put(item)
time.sleep(random.randint(2, 4))
def consumer(cid):
while True:
item = q.get()
print(f'[🙋 consumer {cid}] got {item}')
time.sleep(1)
for i in (1, 2):
threading.Thread(target=producer, args=(i,), daemon=True).start()
threading.Thread(target=consumer, args=(i,), daemon=True).start()q = DiskQueue(path='./', queue_name='testq', cache_size=2)
for i in range(1, 9):
q.put(i)
q.peek(4) # → [1, 2, 3, 4] (items remain in the queue)q.sync() # flush in-memory buffers to disk on demand| Method | Description |
|---|---|
DiskQueue(path, queue_name, cache_size, max_size=None) |
Open or recover a queue rooted at path/queue_name. |
put(obj, block=True) |
Enqueue an object. |
get(block=True) |
Dequeue the next object. |
put_nowait(obj) / get_nowait() |
Non-blocking variants — raise Full / Empty. |
peek(count=1) |
Return the next count items without removing them. |
sync() |
Explicitly flush in-memory buffers to disk. |
len(q) |
Current number of items across memory + disk. |
cd src/tests && pytest -vvvWant to add features, improve existing code, or fix bugs? Fork the repo and open a pull request — issues and PRs are both welcome.
Built as a learning project to explore queue semantics, durability, and concurrency primitives in Python.