Overview

package wittgenstein_job_queue

Wittgenstein Job Queue — retry + dead-letter job queue for platform services.

Classes

• QueueBackend — A retry + dead-letter job queue: enqueue, claim exactly once, resolve.

• PostgresQueueBackend

• Job — A queued unit of work: an arbitrary JSON-serializable payload plus its lifecycle state.

• JobStatus

wittgenstein_job_queue.QueueBackend

mkapi_definition_mkapi class QueueBackend()

Bases : ABC

A retry + dead-letter job queue: enqueue, claim exactly once, resolve.

claim_next is the one method every backend must make safe for competing consumers — N concurrently-running workers calling it against the same queue must never receive the same job. That guarantee (not delivery-to-everyone broadcast) is the whole reason this library exists instead of a Pub/Sub channel.

Methods

• enqueue — Add a new pending job to the queue.

• claim_next — Atomically claim the oldest pending job, or None if the queue is empty.

• mark_done — Mark a claimed job as successfully completed.

• mark_failed — Record a failure.

• is_drained — True once no job on this queue is still pending/claimed.

wittgenstein_job_queue.QueueBackend.enqueue

mkapi_definition_mkapi method QueueBackend.enqueue(queue_name: str, payload: dict[str, Any], *, max_attempts: int = 3) → Job

Add a new pending job to the queue.

wittgenstein_job_queue.QueueBackend.claim_next

mkapi_definition_mkapi method QueueBackend.claim_next(queue_name: str, worker_id: str) → Job | None

Atomically claim the oldest pending job, or None if the queue is empty.

wittgenstein_job_queue.QueueBackend.mark_done

mkapi_definition_mkapi method QueueBackend.mark_done(job_id: str) → None

Mark a claimed job as successfully completed.

wittgenstein_job_queue.QueueBackend.mark_failed

mkapi_definition_mkapi method QueueBackend.mark_failed(job_id: str, error: str) → None

Record a failure.

Requeues to pending (so a future claim_next picks it up again) while attempt_count < max_attempts; once attempts are exhausted the job moves to dead_letter for manual inspection instead of retrying forever.

wittgenstein_job_queue.QueueBackend.is_drained

mkapi_definition_mkapi method QueueBackend.is_drained(queue_name: str) → bool

True once no job on this queue is still pending/claimed.

Lets a producer that enqueues work in dependent phases (e.g. "don't enqueue B until every A job has resolved") poll for phase completion without reaching past this interface into the storage layer.

wittgenstein_job_queue.PostgresQueueBackend

mkapi_definition_mkapi class PostgresQueueBackend(engine: Engine, table_name: str = 'job_queue_message')

Bases : QueueBackend

Methods

• enqueue

• claim_next

• mark_done

• mark_failed

• is_drained

wittgenstein_job_queue.PostgresQueueBackend.enqueue

mkapi_definition_mkapi method PostgresQueueBackend.enqueue(queue_name: str, payload: dict[str, Any], *, max_attempts: int = 3) → Job

wittgenstein_job_queue.PostgresQueueBackend.claim_next

mkapi_definition_mkapi method PostgresQueueBackend.claim_next(queue_name: str, worker_id: str) → Job | None

wittgenstein_job_queue.PostgresQueueBackend.mark_done

mkapi_definition_mkapi method PostgresQueueBackend.mark_done(job_id: str) → None

wittgenstein_job_queue.PostgresQueueBackend.mark_failed

mkapi_definition_mkapi method PostgresQueueBackend.mark_failed(job_id: str, error: str) → None

wittgenstein_job_queue.PostgresQueueBackend.is_drained

mkapi_definition_mkapi method PostgresQueueBackend.is_drained(queue_name: str) → bool

wittgenstein_job_queue.Job

mkapi_definition_mkapi class Job()

Bases : BaseModel

A queued unit of work: an arbitrary JSON-serializable payload plus its lifecycle state.

wittgenstein_job_queue.JobStatus

mkapi_definition_mkapi class JobStatus()

Bases : str, Enum