litequeue

LiteQueue allows you to associate a single stored message with a unique idempotency_key. This feature makes retrying a put() operation safe in cases of 'ambiguous retries' (e.g., when a network connection fails before the producer receives a response).

Key vs. Message Identity

• message_id: A UUIDv7 string that uniquely identifies a specific stored queue message.
• idempotency_key: An optional, case-sensitive string supplied by the producer to identify a single logical submission.

When you retry a put() with the same idempotency_key and the same data, LiteQueue returns the existing message instead of creating a duplicate. This ensures that even if the first attempt actually succeeded, the second attempt won't result in a second message.

# First attempt
message = queue.put(
    '{"account": "A-123", "amount": 10}',
    idempotency_key="charge-2026-0001",
)

# If response is lost, retry with the EXACT same data and key
same_message = queue.put(
    '{"account": "A-123", "amount": 10}',
    idempotency_key="charge-2026-0001",
)

# The message_id will be identical
assert same_message.message_id == message.message_id

In LiteQueue, a claim_id is a unique integer assigned to a specific delivery of a message. While the message_id remains constant for the life of a message, the claim_id increments every time a message is returned to READY and subsequently delivered via pop().

Together, the pair (message_id, claim_id) uniquely identifies a single worker's claim on a message. This mechanism prevents race conditions where a worker might accidentally complete a message that has already been reassigned to another worker due to a timeout or recovery process.

Key properties:

• A new message has a claim_id of 0 (unclaimed).
• The first pop() increments the claim_id to 1.
• Subsequent pop() calls increment the ID further.
• The claim_id is local to a specific message; different messages can share the same ID.
• The claim_id is not an authentication credential; queue access must be controlled separately.

You can chain idempotency across different queues (e.g., an Inbox queue and an Outbox queue) to ensure end-to-end processing stability.

Pattern 1: Derived Keys

An actor can derive an outbox key from the inbox request identifier:

• Inbox Key: request:2026-0001
• Outbox Key: result:request:2026-0001

If the actor runs again, the outbox put() will return the original output message instead of creating a duplicate.

Pattern 2: Using Message IDs

If the inbox has no caller key, use the inbox message_id as the basis for the outbox key:

• Outbox Key: result:<inbox-message-id>

Important Constraints:

• LiteQueue permits only one queue per database file. It does not provide a single transaction across separate inbox and outbox queue files. If you require atomic changes across both, use a different transactional design.
• Keep the output message until the inbox claim and its retry window are complete.

Idempotency keys are tied to the lifecycle of the message they represent.

• Key Reservation: The key is reserved as long as the message exists in the queue.
• Pruning: When queue.prune() is called, the associated idempotency_key is deleted along with the message. Once pruned, that key becomes available again for new, different data.
• Failed Messages: If you use queue.prune(include_failed=False), failed messages are kept, and their keys remain reserved. A replay attempt with that key will return the existing failed message.
• History: LiteQueue does not maintain a permanent history of keys. If you need to track idempotency history indefinitely, you must store those receipts in an external system.

# 1. Put message with key
original = queue.put("old operation", idempotency_key="charge-2026-0001")

# 2. Process and complete message
claim = queue.pop()
queue.done(message_id=claim.message_id, claim_id=claim.claim_id)

# 3. Prune removes the key
queue.prune()

# 4. Key is now free for a new, different operation
replacement = queue.put("new operation", idempotency_key="charge-2026-0001")

Use LiteQueue to manage a persistent task queue backed by SQLite.

1. Initialize: Create a LiteQueue instance by providing a filename (the SQLite database path).
2. Produce: Use .put(data, idempotency_key=None) to add messages. Messages are passed as strings (e.g., JSON).
3. Consume: Use .pop() to retrieve and lock a message. This returns a Message object.
4. Complete: Once processing is finished, call .done(message_id, claim_id) to mark the task as finished. You must provide the claim_id from the popped Message to ensure you aren't completing a stale claim.
5. Cleanup: Use .prune() to remove finished tasks from the database.

from litequeue import LiteQueue

q = LiteQueue(filename="tasks.sqlite3")

# Add messages
q.put("hello", idempotency_key="greeting-1")
q.put("world")

# Pop a message
task = q.pop()
assert task is not None

# Process and mark as done
# Note: task.claim_id is required to prevent stale worker updates
updated = q.done(
    message_id=task.message_id,
    claim_id=task.claim_id,
)
assert updated

To migrate a single queue, use an ATTACH DATABASE command to move data from the source table to a new table named Queue in the destination database. This process must be wrapped in a transaction to ensure atomicity and include the creation of required indexes.

sqlite3 /srv/app/queues.sqlite3 <<'SQL'
.bail on
.timeout 5000
ATTACH DATABASE '/srv/app/new-queues/email.queue.sqlite3' AS target;
BEGIN IMMEDIATE;

CREATE TABLE target."Queue"
(
    data       TEXT NOT NULL,
    message_id TEXT NOT NULL,
    status     INTEGER NOT NULL,
    in_time    INTEGER NOT NULL,
    lock_time  INTEGER,
    done_time  INTEGER
);

INSERT INTO target."Queue"
    (data, message_id, status, in_time, lock_time, done_time)
SELECT
    data, message_id, status, in_time, lock_time, done_time
FROM main."EmailQueue";

CREATE UNIQUE INDEX target."Queue_message_id_unique_idx"
ON "Queue"(message_id);

CREATE INDEX target."Queue_status_message_id_idx"
ON "Queue"(status, message_id);

COMMIT;
DETACH DATABASE target;
SQL