postgres

module wittgenstein_job_queue.backends.postgres

Postgres-backed QueueBackend: FOR UPDATE SKIP LOCKED claim, retry-then-dead-letter.

Operates over a column contract on a table name the caller supplies — this library does not own or migrate that table (schema changes stay the consuming app's own Alembic/migration chain); it only issues SQL against the agreed columns::

id            text/uuid, primary key
queue_name    text
payload       json/jsonb
status        text            (pending | claimed | done | dead_letter)
attempt_count integer
max_attempts  integer
claimed_by    text,           nullable
claimed_at    timestamptz,    nullable
completed_at  timestamptz,    nullable
error         text,           nullable
created_at    timestamptz

The claim query mirrors the FOR UPDATE ... SKIP LOCKED idiom already proven in wittgenstein-core/graphrag-interlinked's pipeline_state reactors — the only addition here is dialect-awareness: SQLite (used by a consuming app's own fast unit tests, no live Postgres) doesn't support SKIP LOCKED, so a plain claim is issued there instead. Concurrency-safety only matters against a real Postgres; the SQLite path exists purely so a consumer's test suite can exercise enqueue/claim/done/failed without a database server.

Classes

• PostgresQueueBackend

wittgenstein_job_queue.backends.postgres.PostgresQueueBackend

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.backends.postgres.PostgresQueueBackend.enqueue

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

wittgenstein_job_queue.backends.postgres.PostgresQueueBackend.claim_next

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

wittgenstein_job_queue.backends.postgres.PostgresQueueBackend.mark_done

method PostgresQueueBackend.mark_done(job_id: str) → None

wittgenstein_job_queue.backends.postgres.PostgresQueueBackend.mark_failed

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

wittgenstein_job_queue.backends.postgres.PostgresQueueBackend.is_drained

method PostgresQueueBackend.is_drained(queue_name: str) → bool