Queue Backends#

Registry#

Queue backends are registered through litestar_queues.backends.

litestar_queues.backends.get_queue_backend(backend: QueueBackendConfig = 'memory', config: QueueConfig | None = None) → BaseQueueBackend[source]#

Get an instantiated queue backend.

Returns:: A configured queue backend instance.
Raises:: TypeError – If a typed backend config selects a backend class that does not accept backend_config.

litestar_queues.backends.get_queue_backend_class(backend_path: str) → type[BaseQueueBackend][source]#

Get a queue backend class by short name or import path.

Optional backends are imported lazily on first lookup so unused adapters do not require their driver extras to be installed.

Returns:: The resolved queue backend class.
Raises:: ValueError – If a short backend name is unknown.

litestar_queues.backends.queue_backend(name: str) → Callable[[type[BaseQueueBackend]], type[BaseQueueBackend]][source]#

Decorator to register a queue backend class with a short name.

Returns:: A decorator that registers the backend class.

litestar_queues.backends.list_queue_backends() → list[str][source]#: Return registered queue backend names (built-ins + dynamically registered).

Base Backend#

class litestar_queues.backends.base.BaseQueueBackend(config: QueueConfig | None = None)[source]#

Bases: object

Base class for queue persistence backends.

__init__(config: QueueConfig | None = None) → None[source]#: Initialize the queue backend.

config#

property capabilities: QueueBackendCapabilities#: Backend behavior capabilities.

async open() → bool[source]#

Open queue resources.

Returns:: True when resources are ready.

async close() → None[source]#: Close queue resources.

get_event_log(config: EventHistoryConfig) → QueueEventLog | None[source]#: Return a backend-owned queue event history implementation, if supported.

async enqueue(task_name: str, *, args: tuple[Any, ...] = (), kwargs: dict[str, Any] | None = None, queue: str = 'default', priority: int = 0, max_retries: int = 0, scheduled_at: datetime | None = None, expires_at: datetime | None = None, key: str | None = None, execution_backend: str = 'local', execution_profile: str | None = None, metadata: dict[str, Any] | None = None, id: UUID | None = None) → QueuedTaskRecord[source]#

Persist a queued task.

When id is provided the persisted record uses it instead of a freshly generated identifier; the service pre-generates it for unique_until="forever" enqueues so the identity reservation and the executable record share one id.

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist multiple queued tasks, returning records in input order.

The default implementation issues one enqueue() per request, which preserves per-key deduplication and ordering. Backends with a native bulk path (e.g. SQLSpec COPY/Arrow/execute_many) override this for throughput while keeping the same semantics.

Returns:: Queue task records in the same order as requests.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#: Return a queued task by ID.

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#: Return a queued task by deduplication key.

async get_tasks(task_ids: Sequence[UUID]) → list[QueuedTaskRecord][source]#: Return existing records for the supplied identifiers.

async notify_worker_control(worker_id: str | None) → None[source]#

Publish a best-effort worker-control hint.

worker_id is the record’s persisted owner, or None when the record was cancelled before any worker claimed it. It travels for observability only: the control channel is shared and every subscribed worker reconciles its own running tasks against durable status on receipt.

Worker-control hints are lossy: durable status remains authoritative, and a dropped hint costs cancellation latency, never correctness.

async wait_for_worker_control(*, worker_id: str, timeout: float | None = None) → bool[source]#

Wait for a best-effort worker-control hint.

Mirrors wait_for_wakeups(): polling-only backends inherit this sleep-and-report-nothing default and stay pure-poll.

Returns:: True when a control hint was observed.

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist the owner of a running retry generation.

The write is fenced on status='running' AND retry_count = expected so a worker can never claim ownership of a generation it lost.

Parameters:

task_id – Queue record identifier.
worker_id – Identity to persist as the record owner.
expected_retry_count – Retry generation the caller believes it owns.

Raises:

NotImplementedError – Always; every backend must answer this.

async list_pending(*, limit: int = 1, queue: str | None = None, execution_backend: str | None = None) → list[QueuedTaskRecord][source]#: Return due pending or scheduled tasks ordered for execution.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#: Atomically claim a pending task.

async claim_task_with_expired(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → tuple[QueuedTaskRecord | None, QueuedTaskRecord | None][source]#

Claim one task and report when this call expires that task.

Returns:: The claimed record and the expired record, at most one of which is set.

async claim_next(*, queues: tuple[str, ...] = (), execution_backend: str | None = None) → QueuedTaskRecord | None[source]#

Claim the next due task across the requested queues.

An empty queues tuple claims across all queues.

Returns:: The claimed task record, if one was available.

async claim_many(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → list[QueuedTaskRecord][source]#

Claim up to limit due tasks across the requested queues.

An empty queues tuple claims across all queues. Backends with a native batch-claim primitive override this method; the fallback here preserves claim_next() semantics for backends with only a single-record primitive.

Returns:: Claimed task records.

async claim_many_with_expired(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#

Claim records and report overdue records transitioned while claiming.

Returns:: Claimed records and records expired by this call.

async complete_task(task_id: UUID, *, result: Any = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Mark a task as completed.

Parameters:

task_id – Queue record identifier.
result – Task result payload.
expected_retry_count – When provided, update only if the record is still running with this retry count.

async fail_task(task_id: UUID, error: str, *, retry: bool = True, expected_retry_count: int | None = None, retry_at: datetime | None = None, queued_at: datetime | None = None) → QueuedTaskRecord | None[source]#

Mark a task as failed or retry it.

Parameters:

task_id – Queue record identifier.
error – Error message to persist.
retry – Whether retry policy may requeue the task.
expected_retry_count – When provided, update only if the record is still running with this retry count.
retry_at – Scheduled eligibility timestamp for a delayed retry.
queued_at – Queue ordering timestamp for the new retry attempt.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

Cancel a task.

Parameters:

task_id – ID of the task to cancel.
include_running – Whether to cancel a task that is currently running.
expected_retry_count – Only cancel if the task is still on this retry generation.

Returns:

True if the task was successfully cancelled, False otherwise.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending after interruption.

The write is fenced on status='running' AND retry_count = expected AND worker_id = worker_id. It resets the record to pending with a fresh queued_at and clears scheduled_at, started_at, heartbeat_at, completed_at, execution_ref, and worker_id.

Parameters:

task_id – Queue record identifier.
expected_retry_count – Retry generation the caller owns.
worker_id – Identity that must currently own the record.
queued_at – Requeue timestamp used for fair claim ordering.

Raises:

NotImplementedError – Always; every backend must answer this.

async cancel_tasks(*, task_name: str | None = None, queue: str | None = None, kwargs: Mapping[str, Any] | None = None, metadata: Mapping[str, Any] | None = None, include_running: bool = False) → int[source]#

Cancel tasks matching a domain predicate.

Parameters:

task_name – Optional task name exact match.
queue – Optional queue exact match.
kwargs – Optional top-level kwargs exact-match subset.
metadata – Optional top-level metadata exact-match subset.
include_running – When true, running records are included for cooperative cancellation.

Returns:

Number of records cancelled.

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

Update heartbeat timestamps for running tasks.

Returns:: The task IDs confirmed touched or missed by the backend.
Raises:: NotImplementedError – Always; every backend must answer this.

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#

Clear heartbeat timestamps for task IDs.

Parameters:

task_ids – Queue record identifiers.
expected_retry_count – When provided, clear only records that still match this retry count.

Raises:

NotImplementedError – Always; every backend must answer this.

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None) → StaleTaskRecoveryResult[source]#

Recover running tasks with stale heartbeats.

Parameters:

stale_after – Heartbeat age past which a running task is stale.
limit – When provided, recover at most this many records ordered oldest-first (oldest heartbeat, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

Summary of requeued, failed, skipped, and handler-needed records.

Raises:

NotImplementedError – Always; every backend must answer this.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Transition overdue pending or scheduled records to expired.

Returns:: Records transitioned to expired.
Raises:: NotImplementedError – Always; every backend must answer this.

async acquire_worker_lock(name: str, *, ttl: timedelta) → bool[source]#

Acquire a backend-scoped worker coordination lock.

Fleet coordination and maintenance ownership are the same primitive, so this routes to acquire_maintenance() under a fresh token. The lock is never released explicitly: ttl bounds it so a worker that dies mid-pass cannot wedge the fleet, and the next holder takes over once the stored ownership expires.

Returns:: True when the caller should run the coordinated worker action.

async acquire_maintenance(name: str, token: str, *, ttl: timedelta) → bool[source]#

Acquire token-fenced distributed maintenance ownership.

Only backends advertising supports_maintenance implement a real coordination record. The base raises so maintenance fails closed rather than silently running unfenced on a backend that cannot prevent overlapping runs.

Raises:: NotImplementedError – Always, on backends without maintenance support.

async release_maintenance(name: str, token: str) → bool[source]#

Release maintenance ownership held under token.

Releases only when the persisted token matches token, so a stale holder can never delete a successor’s ownership record.

Returns:: True when ownership held under token was released.
Raises:: NotImplementedError – Always, on backends without maintenance support.

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an external execution reference for a running task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async reserve_external_dispatch(task_id: UUID, execution_backend: str, reservation_ref: str, *, execution_profile: str | None = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Atomically reserve a due, unexpired task for external dispatch.

The default rejects dispatch because a read-then-write fallback cannot protect the external side effect.

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#: Clear an exact pending external attempt and wake dispatchers.

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#: Atomically rotate an exact pending external attempt reference.

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Release a matching external-dispatch reservation.

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Replace an owned dispatch reservation with its execution reference.

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an execution backend/profile change for a queued task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Return externally dispatched tasks with references to reconcile.

Raises:: NotImplementedError – Always; every backend must answer this.

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#

Return queue status counts.

Raises:: NotImplementedError – Always; every backend must answer this.

async list_completed_by_task(task_name: str, *, since: datetime | None = None, limit: int = 10) → list[QueuedTaskRecord][source]#

Return recent completed records for a task name.

Raises:: NotImplementedError – Always; every backend must answer this.

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

Delete terminal records completed before a cutoff.

Routine terminal cleanup never touches unique_until="forever" reservations; only reset_identity() removes them.

Parameters:

before – Delete terminal records completed strictly before this UTC cutoff.
limit – When provided, delete at most this many records ordered oldest-first (oldest completed_at, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

The number of deleted records.

Raises:

NotImplementedError – Always; every backend must answer this.

async reserve_identity(key: str, *, task_id: UUID, task_name: str) → TaskReservation | None[source]#

Atomically reserve a unique_until="forever" identity.

Reservation is atomic: exactly one concurrent caller wins a given key. The winner receives None and owns the durable reservation; every other caller receives the existing owner reservation. Reservation is the only way a reservation is created and must run before the executable record is persisted so a committed forever task can never lack its reservation.

Parameters:

key – The effective identity key to reserve.
task_id – The originating task id (shared with the executable record).
task_name – The originating registered task name.

Returns:

None when this caller won the reservation; otherwise the existing owner reservation.

async has_identity(key: str) → TaskReservation | None[source]#: Return the reservation owning a reserved forever identity, if any.

async reset_identity(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete a forever identity reservation.

This is the only reservation deletion path; routine terminal and event maintenance never remove reservations. When expected_task_id is provided, delete only when that task still owns the reservation. This compare-and-delete form lets enqueue recovery release its own failed reservation without deleting a successor created after an explicit reset. Omitting it preserves the explicit administrative reset behavior.

Parameters:

key – The exact effective identity key.
expected_task_id – Optional task owner required for deletion.

Returns:

True when a reservation was removed.

async notify_new_task(record: QueuedTaskRecord) → None[source]#: Notify waiters that a new task is available.

async notify_new_tasks(records: Sequence[QueuedTaskRecord]) → None[source]#: Emit one worker-wakeup hint for a batch of newly available tasks.

async wait_for_wakeups(timeout: float | None = None) → bool[source]#

Wait until backend notification arrives.

Returns:: True when a notification was observed.

async time_until_next_due(*, queues: tuple[str, ...] = ()) → float | None[source]#

Return seconds until the earliest not-yet-due pending/scheduled record.

Bounds the worker’s adaptive polling wait so a scheduled or retried task is never discovered later than its own due time: no backend has a push notification for “a record’s scheduled time arrived,” so a worker asleep on a long backoff wait would otherwise only notice after that wait elapses. The default reports None (unknown); concrete backends that can answer this cheaply override it. An unfiltered or slightly-early answer is always safe here (it can only wake the worker sooner than strictly necessary, never later).

Returns:: Seconds until the next due record across queues (all queues when empty), or None when there is no upcoming scheduled work or the backend does not support this query.

async wait_for_completion(task_id: UUID, *, timeout: float | None = None) → bool[source]#

Wait for a terminal-completion signal for one task.

Backends that advertise supports_completion_events override this to subscribe to a completion channel. The default returns False so callers fall back to polling.

Returns:: True when a completion signal for task_id was observed.

litestar_queues.backends.base.attempts_consumed(record: QueuedTaskRecord) → int[source]#

Return the retry attempts a record has actually consumed.

An interruption bumps retry_count so the old owner’s fences can never settle the reclaimed attempt, but it is not a failed attempt: it must not spend the record’s retry budget.

Returns:: retry_count less the recorded interruptions.

litestar_queues.backends.base.interruption_count(record: QueuedTaskRecord) → int[source]#

Return how many times an attempt has been interrupted by a shutdown.

Returns:: The recorded interruption count, or 0 when it is absent or unusable.

litestar_queues.backends.base.is_external_dispatch_reservation(execution_ref: str | None) → bool[source]#: Return whether an execution reference is a temporary dispatch lease.

litestar_queues.backends.base.retry_schedule(record: QueuedTaskRecord, *, now: datetime | None = None) → tuple[datetime, datetime | None][source]#: Return refreshed queue and optional due timestamps for a retry.

Memory#

class litestar_queues.backends.memory.backend.InMemoryQueueBackend(config: QueueConfig | None = None)[source]#

Bases: BaseQueueBackend

In-process queue backend for tests, local development, and examples.

__init__(config: QueueConfig | None = None) → None[source]#: Initialize the queue backend.

property capabilities: QueueBackendCapabilities#: Backend behavior capabilities.

get_event_log(config: EventHistoryConfig) → QueueEventLog | None[source]#: Return bounded, process-local queue event history when enabled.

Persist a queued task.

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist multiple in-memory tasks while signaling waiters once.

Returns:: Queue task records in the same order as requests.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#: Return a queued task by ID.

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#: Return a queued task by deduplication key.

async get_tasks(task_ids: Sequence[UUID]) → list[QueuedTaskRecord][source]#: Return existing records for the supplied identifiers.

async notify_worker_control(worker_id: str | None) → None[source]#

Publish a best-effort worker-control hint.

Worker-control hints are lossy: durable status remains authoritative, and a dropped hint costs cancellation latency, never correctness.

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist the owner of a running retry generation.

The write is fenced on status='running' AND retry_count = expected so a worker can never claim ownership of a generation it lost.

Parameters:

task_id – Queue record identifier.
worker_id – Identity to persist as the record owner.
expected_retry_count – Retry generation the caller believes it owns.

Raises:

NotImplementedError – Always; every backend must answer this.

async list_pending(*, limit: int = 1, queue: str | None = None, execution_backend: str | None = None) → list[QueuedTaskRecord][source]#: Return due pending or scheduled tasks ordered for execution.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#: Atomically claim a pending task.

async claim_task_with_expired(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → tuple[QueuedTaskRecord | None, QueuedTaskRecord | None][source]#: Claim one task and return an expiry transitioned under the same lock.

async claim_many(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → list[QueuedTaskRecord][source]#

Claim up to limit due tasks under a single lock acquisition.

Selects eligible records with the same queue/execution/due filter and priority ordering as list_pending(), then transitions them to running inside one critical section using a single now snapshot. The returned records carry the same owner/start/heartbeat fields a sequential claim_next() loop would produce.

Returns:: Claimed task records in claim order.

async claim_many_with_expired(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#: Claim records and return overdue records expired under the same lock.

async complete_task(task_id: UUID, *, result: Any = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Mark a task as completed.

Parameters:

task_id – Queue record identifier.
result – Task result payload.
expected_retry_count – When provided, update only if the record is still running with this retry count.

Mark a task as failed or retry it.

Parameters:

task_id – Queue record identifier.
error – Error message to persist.
retry – Whether retry policy may requeue the task.
expected_retry_count – When provided, update only if the record is still running with this retry count.
retry_at – Scheduled eligibility timestamp for a delayed retry.
queued_at – Queue ordering timestamp for the new retry attempt.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

Cancel a task.

Parameters:

task_id – ID of the task to cancel.
include_running – Whether to cancel a task that is currently running.
expected_retry_count – Only cancel if the task is still on this retry generation.

Returns:

True if the task was successfully cancelled, False otherwise.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending after interruption.

Parameters:

task_id – Queue record identifier.
expected_retry_count – Retry generation the caller owns.
worker_id – Identity that must currently own the record.
queued_at – Requeue timestamp used for fair claim ordering.

Raises:

NotImplementedError – Always; every backend must answer this.

Cancel tasks matching a domain predicate.

Parameters:

task_name – Optional task name exact match.
queue – Optional queue exact match.
kwargs – Optional top-level kwargs exact-match subset.
metadata – Optional top-level metadata exact-match subset.
include_running – When true, running records are included for cooperative cancellation.

Returns:

Number of records cancelled.

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

Update heartbeat timestamps for running tasks.

Returns:: The task IDs confirmed touched or missed by the backend.
Raises:: NotImplementedError – Always; every backend must answer this.

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#

Clear heartbeat timestamps for task IDs.

Parameters:

task_ids – Queue record identifiers.
expected_retry_count – When provided, clear only records that still match this retry count.

Raises:

NotImplementedError – Always; every backend must answer this.

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None) → StaleTaskRecoveryResult[source]#

Recover running tasks with stale heartbeats.

Parameters:

stale_after – Heartbeat age past which a running task is stale.
limit – When provided, recover at most this many records ordered oldest-first (oldest heartbeat, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

Summary of requeued, failed, skipped, and handler-needed records.

Raises:

NotImplementedError – Always; every backend must answer this.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Transition overdue pending or scheduled records to expired.

Returns:: Records transitioned during this call.

Atomically reserve a due, unexpired task for external dispatch.

The default rejects dispatch because a read-then-write fallback cannot protect the external side effect.

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#: Clear an exact pending external attempt and wake dispatchers.

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#: Atomically rotate an exact pending external attempt reference.

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Release a matching external-dispatch reservation.

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Replace an owned dispatch reservation with its execution reference.

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an external execution reference for a running task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an execution backend/profile change for a queued task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Return externally dispatched tasks with references to reconcile.

Raises:: NotImplementedError – Always; every backend must answer this.

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#

Return queue status counts.

Raises:: NotImplementedError – Always; every backend must answer this.

async list_completed_by_task(task_name: str, *, since: datetime | None = None, limit: int = 10) → list[QueuedTaskRecord][source]#

Return recent completed records for a task name.

Raises:: NotImplementedError – Always; every backend must answer this.

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

Delete terminal records completed before a cutoff.

Routine terminal cleanup never touches unique_until="forever" reservations; only reset_identity() removes them.

Parameters:

before – Delete terminal records completed strictly before this UTC cutoff.
limit – When provided, delete at most this many records ordered oldest-first (oldest completed_at, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

The number of deleted records.

Raises:

NotImplementedError – Always; every backend must answer this.

async acquire_maintenance(name: str, token: str, *, ttl: timedelta) → bool[source]#

Acquire expiring, token-fenced maintenance ownership under the async lock.

Returns:: True when ownership was granted to token.

async release_maintenance(name: str, token: str) → bool[source]#

Release maintenance ownership only when token matches the holder.

Returns:: True when ownership held under token was released.

async reserve_identity(key: str, *, task_id: UUID, task_name: str) → TaskReservation | None[source]#

Reserve a forever identity under the shared lock beside key ownership.

Returns:: None when this caller won the reservation; otherwise the existing owner reservation.

async has_identity(key: str) → TaskReservation | None[source]#: Return the reservation owning a reserved forever identity, if any.

async reset_identity(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete a forever identity reservation under the shared lock.

Parameters:

key – The exact effective identity key.
expected_task_id – Optional task owner required for deletion.

Returns:

True when a reservation was removed.

async notify_new_task(record: QueuedTaskRecord) → None[source]#: Notify waiters that a new task is available.

async wait_for_wakeups(timeout: float | None = None) → bool[source]#

Wait until backend notification arrives.

Returns:: True when a notification was observed.

async wait_for_worker_control(*, worker_id: str, timeout: float | None = None) → bool[source]#

Wait for a best-effort worker-control hint.

Mirrors wait_for_wakeups(): polling-only backends inherit this sleep-and-report-nothing default and stay pure-poll.

Returns:: True when a control hint was observed.

async time_until_next_due(*, queues: tuple[str, ...] = ()) → float | None[source]#

Return seconds until the earliest not-yet-due pending/scheduled record.

Returns:: Seconds until the next due record, or None when there is no upcoming scheduled work.

async close() → None[source]#: Cancel any retained notification and worker-control waits.

async clear() → None[source]#: Clear all in-memory records.

In-memory queue event history.

class litestar_queues.backends.memory.event_log.InMemoryQueueEventLog(config: EventHistoryConfig)[source]#

Bases: object

Process-local, bounded queue event history for tests and local usage.

__init__(config: EventHistoryConfig) → None[source]#

async publish_event(event: QueueEvent) → None[source]#: Append an event history record and prune the oldest records.

async flush_events() → None[source]#

Flush buffered events.

The memory event log writes immediately, so this is intentionally a no-op.

async query_events(query: QueueEventQuery | None = None, *, extra: Mapping[str, str] | None = None) → OffsetPagination[QueueEventLogRecord][source]#

Return a filtered, ordered page of event history records.

Returns:: The matching page.

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary][source]#

Return per-stage aggregates for the matching records.

Raises:: QueueConfigurationError – If query sets ordering or pagination.
Returns:: One summary per distinct stage.

async cleanup_events(*, before: datetime, match: QueueEventQuery | None = None, exclude: Sequence[QueueEventQuery] = (), limit: int | None = None) → int[source]#

Delete the oldest matching records occurring before before.

A record is deleted only when it matches match and matches none of exclude. Deletion order is the ascending stable order key, so repeated bounded calls converge.

Returns:: Number of deleted records.

async clear() → None[source]#: Clear all memory event-history records.

SQLSpec#

SQLSpec queue backend.

class litestar_queues.backends.sqlspec.backend.SQLSpecQueueBackend(config: QueueConfig | None = None, *, backend_config: SQLSpecBackendConfig | None = None)[source]#

Bases: BaseQueueBackend

SQLSpec-backed queue backend.

__init__(config: QueueConfig | None = None, *, backend_config: SQLSpecBackendConfig | None = None) → None[source]#: Initialize the queue backend.

async open() → bool[source]#

Open SQLSpec resources.

Returns:: True when SQLSpec resources are ready.

async close() → None[source]#: Close SQLSpec resources.

get_event_log(config: EventHistoryConfig) → QueueEventLog | None[source]#: Return SQLSpec-managed durable queue event history when enabled.

property capabilities: QueueBackendCapabilities#: Backend behavior capabilities.

async create_schema() → None[source]#: Create the SQLSpec queue table and indexes.

Persist a queued task.

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist many tasks via the adapter’s fastest bulk path.

Resolves existing deduplication keys in one round trip, then inserts the remaining rows through the native Arrow ingest path (load_from_records()) when the adapter supports it, otherwise via a batched execute_many. Returns records in input order, with existing non-terminal keyed tasks returned as-is (no duplicate insert) to match the semantics of enqueue().

Returns:: Queue task records in the same order as requests.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#: Return a queued task by ID.

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#: Return a queued task by deduplication key.

async reserve_identity(key: str, *, task_id: UUID, task_name: str) → TaskReservation | None[source]#

Reserve a forever identity via an optimistic insert with a unique-violation fallback.

The reservation table’s identity-key PRIMARY KEY is the atomicity arbiter: exactly one concurrent insert wins; a loser catches the unique violation and reads the winning owner. Serializable engines (CockroachDB) may abort a losing transaction with a serialization/retry error before the unique violation surfaces, so those are retried with bounded backoff. The reservation table is separate from the queue table and terminal cleanup never touches it.

Returns:: None when this caller won the reservation; otherwise the existing owner reservation.

async has_identity(key: str) → TaskReservation | None[source]#: Return the reservation owning a reserved forever identity, if any.

async reset_identity(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete a forever identity reservation via count-then-delete.

Returns:: True when a reservation was removed.

async list_pending(*, limit: int = 1, queue: str | None = None, execution_backend: str | None = None) → list[QueuedTaskRecord][source]#: Return due pending or scheduled tasks ordered for execution.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#: Atomically claim a pending task.

async claim_task_with_expired(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → tuple[QueuedTaskRecord | None, QueuedTaskRecord | None][source]#: Claim one task and identify a claim-time expiry owned by this call.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#: Transition overdue pending or scheduled records to expired.

async claim_many(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → list[QueuedTaskRecord][source]#

Claim up to limit due tasks.

Postgres-family stores use one autocommit UPDATE ... FROM (SELECT ... FOR UPDATE SKIP LOCKED LIMIT $n) ... RETURNING statement. Other adapters fall back to the per-record claim loop.

Returns:: Claimed task records.

async claim_many_with_expired(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#: Claim records and report every expiry transition owned by this call.

async claim_next(*, queues: tuple[str, ...] = (), execution_backend: str | None = None) → QueuedTaskRecord | None[source]#

Claim the next due task across the requested queues.

An empty queues tuple claims across all queues.

Returns:: The claimed task record, if one was available.

async complete_task(task_id: UUID, *, result: Any | None = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Mark a task as completed.

Parameters:

task_id – Queue record identifier.
result – Task result payload.
expected_retry_count – When provided, update only if the record is still running with this retry count.

Mark a task as failed or retry it.

Parameters:

task_id – Queue record identifier.
error – Error message to persist.
retry – Whether retry policy may requeue the task.
expected_retry_count – When provided, update only if the record is still running with this retry count.
retry_at – Scheduled eligibility timestamp for a delayed retry.
queued_at – Queue ordering timestamp for the new retry attempt.

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist the owner of a running retry generation.

The write is fenced on status='running' AND retry_count = expected so a worker can never claim ownership of a generation it lost.

Parameters:

task_id – Queue record identifier.
worker_id – Identity to persist as the record owner.
expected_retry_count – Retry generation the caller believes it owns.

Raises:

NotImplementedError – Always; every backend must answer this.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending after interruption.

Parameters:

task_id – Queue record identifier.
expected_retry_count – Retry generation the caller owns.
worker_id – Identity that must currently own the record.
queued_at – Requeue timestamp used for fair claim ordering.

Raises:

NotImplementedError – Always; every backend must answer this.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

Cancel a task.

Parameters:

task_id – ID of the task to cancel.
include_running – Whether to cancel a task that is currently running.
expected_retry_count – Only cancel if the task is still on this retry generation.

Returns:

True if the task was successfully cancelled, False otherwise.

Cancel tasks matching a domain predicate.

Parameters:

task_name – Optional task name exact match.
queue – Optional queue exact match.
kwargs – Optional top-level kwargs exact-match subset.
metadata – Optional top-level metadata exact-match subset.
include_running – When true, running records are included for cooperative cancellation.

Returns:

Number of records cancelled.

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

Update heartbeat timestamps for running tasks.

Returns:: The task IDs confirmed touched or missed by the backend.
Raises:: NotImplementedError – Always; every backend must answer this.

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#

Clear heartbeat timestamps for task IDs.

Parameters:

task_ids – Queue record identifiers.
expected_retry_count – When provided, clear only records that still match this retry count.

Raises:

NotImplementedError – Always; every backend must answer this.

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None) → StaleTaskRecoveryResult[source]#

Recover running tasks with stale heartbeats.

Parameters:

stale_after – Heartbeat age past which a running task is stale.
limit – When provided, recover at most this many records ordered oldest-first (oldest heartbeat, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

Summary of requeued, failed, skipped, and handler-needed records.

Raises:

NotImplementedError – Always; every backend must answer this.

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an external execution reference for a running task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

Atomically reserve a due, unexpired task for external dispatch.

The default rejects dispatch because a read-then-write fallback cannot protect the external side effect.

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#: Clear an exact pending external attempt and wake dispatchers.

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#: Atomically rotate an exact pending external attempt reference.

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Release a matching external-dispatch reservation.

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Replace an owned dispatch reservation with its execution reference.

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an execution backend/profile change for a queued task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Return externally dispatched tasks with references to reconcile.

Raises:: NotImplementedError – Always; every backend must answer this.

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#

Return queue status counts.

Raises:: NotImplementedError – Always; every backend must answer this.

async iter_all(*, chunk_size: int = 1000) → AsyncIterator[QueuedTaskRecord][source]#

Stream every queue record without materializing the full table.

Uses SQLSpec select_stream so large administrative scans and exports consume rows in chunks of chunk_size rather than loading the entire result set into memory. The backend session stays open for the duration of iteration, so callers should consume the iterator promptly.

Yields:: Queue task records from the backing SQLSpec table.

async list_completed_by_task(task_name: str, *, since: datetime | None = None, limit: int = 10) → list[QueuedTaskRecord][source]#

Return recent completed records for a task name.

Raises:: NotImplementedError – Always; every backend must answer this.

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

Delete terminal records completed before a cutoff.

Routine terminal cleanup never touches unique_until="forever" reservations; only reset_identity() removes them.

Parameters:

before – Delete terminal records completed strictly before this UTC cutoff.
limit – When provided, delete at most this many records ordered oldest-first (oldest completed_at, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

The number of deleted records.

Raises:

NotImplementedError – Always; every backend must answer this.

async acquire_maintenance(name: str, token: str, *, ttl: timedelta) → bool[source]#

Acquire maintenance ownership via an adapter-portable compare-and-set.

Updates the named row to this token when its stored ownership has expired; if a live row is present the update matches nothing and the row’s token differs, so ownership is denied. When no row exists a fresh one is inserted, and a uniqueness race is treated as ownership denial. The select-after-update read makes correctness independent of the adapter’s rows_affected reliability.

Returns:: True when maintenance ownership is held under token.

async release_maintenance(name: str, token: str) → bool[source]#

Release maintenance ownership only when the stored token matches.

Returns:: True when ownership held under token was deleted and no successor replaced it before the transaction’s postcondition check.

async notify_new_task(record: QueuedTaskRecord) → None[source]#: Publish a SQLSpec event when configured queue work becomes available.

async wait_for_wakeups(timeout: float | None = None) → bool[source]#

Wait for a SQLSpec event when queue notifications are configured.

One iter_events stream and its pending anext read are retained across worker poll timeouts; only an event, a driver failure, or backend close ends them.

Returns:: True when a notification was received.

async notify_worker_control(worker_id: str | None) → None[source]#

Publish a worker-control hint on the SQLSpec events channel.

On the Postgres drivers this is a LISTEN/NOTIFY message. The hint is lossy by contract: it only shortens the wait before the owning worker reconciles durable status.

async wait_for_worker_control(*, worker_id: str, timeout: float | None = None) → bool[source]#

Wait for a SQLSpec worker-control hint.

One iter_events stream and its pending read are retained across worker poll timeouts, exactly like the wakeup stream, and are never shared with it.

Returns:: True when a control hint was observed.
Raises:: Exception – Whatever the event read raised, after the stream is closed so the next wait re-establishes it.

async time_until_next_due(*, queues: tuple[str, ...] = ()) → float | None[source]#

Return seconds until the earliest not-yet-due pending/scheduled record.

Returns:: Seconds until the next due record, or None when there is no upcoming scheduled work.

SQLSpec backend configuration.

litestar_queues.backends.sqlspec.config.WAKEUP_TRANSPORTS: frozenset[str] = frozenset({'aq', 'notify', 'notify_queue', 'poll_queue', 'polling', 'txeventq'})#

Valid worker-wakeup transports for SQLSpecWorkerWakeupConfig.transport.

notify uses native push wakeups, notify_queue uses native push wakeups with a durable queue fallback, poll_queue uses the durable events table, aq and txeventq use Oracle Advanced Queuing backends, and polling disables push wakeups so workers fall back to interval polling.

class litestar_queues.backends.sqlspec.config.SQLSpecBackendConfig(sqlspec: SQLSpec | None = None, sqlspec_config: SQLSpecStoreConfig | None = None, heartbeat_pool_config: SQLSpecStoreConfig | None = None, queue_table_name: str | None = None, worker_wakeups: SQLSpecWorkerWakeupConfig | None = <factory>, event_history_table_name: str | None = None, event_history_extra_columns: tuple[EventHistoryExtraColumn, ...] = (), maintenance_table_name: str | None = None, task_reservation_table_name: str | None = None, column_map: Mapping[str, str] = <factory>, native_json_columns: frozenset[str] = <factory>, manage_schema: bool = True)[source]#

Bases: object

Configuration values for the SQLSpec queue backend.

backend_name: ClassVar[str] = 'sqlspec'#

sqlspec: SQLSpec | None#: Injected SQLSpec manager; None creates a manager owned by the queue backend.

sqlspec_config: SQLSpecStoreConfig | None#: SQLSpec adapter configuration used for queue operations; None resolves or creates one.

heartbeat_pool_config: SQLSpecStoreConfig | None#: Dedicated heartbeat adapter configuration; None reuses normal queue operations.

queue_table_name: str | None#: Queue-task table name; None uses queue_task or SQLSpec extension settings.

__init__(sqlspec: SQLSpec | None = None, sqlspec_config: SQLSpecStoreConfig | None = None, heartbeat_pool_config: SQLSpecStoreConfig | None = None, queue_table_name: str | None = None, worker_wakeups: SQLSpecWorkerWakeupConfig | None = <factory>, event_history_table_name: str | None = None, event_history_extra_columns: tuple[EventHistoryExtraColumn, ...] = (), maintenance_table_name: str | None = None, task_reservation_table_name: str | None = None, column_map: Mapping[str, str] = <factory>, native_json_columns: frozenset[str] = <factory>, manage_schema: bool = True) → None#

worker_wakeups: SQLSpecWorkerWakeupConfig | None#: Worker wakeup transport configuration; None disables wakeups.

event_history_table_name: str | None#: Task-event history table name; None derives it from the queue-task table.

event_history_extra_columns: tuple[EventHistoryExtraColumn, ...]#: Adopter-declared extra scoping columns on the event-history table.

maintenance_table_name: str | None#: Maintenance coordination table name; None derives the package default.

task_reservation_table_name: str | None#: Permanent task-reservation table name; None derives it from the queue-task table.

column_map: Mapping[str, str]#: Overrides mapping logical queue fields to adopter-owned database columns.

native_json_columns: frozenset[str]#: Logical queue fields stored in database-native JSON columns.

manage_schema: bool#: Whether backend startup and migrations may create package-owned queue tables.

configure_migrations(config: QueueConfig) → None[source]#

Implements MigrationConfiguringBackend, so QueuePlugin reaches this without importing the SQLSpec backend.

class litestar_queues.backends.sqlspec.config.SQLSpecWorkerWakeupConfig(channel: AsyncEventChannel | None = None, transport: str | None = None, channel_name: str | None = None, queue_table_name: str | None = None, poll_interval: float | None = None, settings: dict[str, Any] = <factory>)[source]#

Bases: object

SQLSpec worker-wakeup channel and transport configuration.

channel: AsyncEventChannel | None#: Explicit SQLSpec event channel; None constructs one from configuration.

transport: str | None#: Explicit wakeup transport; None selects the adapter capability.

channel_name: str | None#: Logical worker-wakeup channel name; None uses the package default.

queue_table_name: str | None#: Durable SQLSpec event-queue table; None derives it from the adapter.

poll_interval: float | None#: SQLSpec event-store poll interval in seconds; None uses SQLSpec defaults.

settings: dict[str, Any]#: Additional SQLSpec events-extension settings.

__init__(channel: AsyncEventChannel | None = None, transport: str | None = None, channel_name: str | None = None, queue_table_name: str | None = None, poll_interval: float | None = None, settings: dict[str, Any] = <factory>) → None#

Schema and migration helpers for the SQLSpec queue backend.

litestar_queues.backends.sqlspec.schema.EVENT_HISTORY_COLUMNS = ('event_id', 'event_type', 'task_id', 'task_name', 'queue', 'worker_id', 'execution_backend', 'execution_profile', 'actor_type', 'actor_id', 'stage', 'level', 'message', 'detail', 'progress_current', 'progress_total', 'progress_percent', 'duration_ms', 'sequence', 'occurred_at', 'created_at', 'scope', 'scope_key', 'actor', 'entity')#: Physical columns the package owns on the SQLSpec event-history table.

litestar_queues.backends.sqlspec.schema.event_history_table_name_for(table_name: str) → str[source]#

Return the default event-history table for a queue table name.

Schema-qualified names keep their schema and append DEFAULT_EVENT_HISTORY_TABLE_SUFFIX to the table part.

litestar_queues.backends.sqlspec.schema.maintenance_table_name_for(table_name: str) → str[source]#

Return the default maintenance table for a queue table name.

Schema-qualified names keep their schema and append DEFAULT_MAINTENANCE_TABLE_SUFFIX to the table part. When the derived name would exceed the portable identifier limit it is deterministically shortened (truncated table part plus a stable hash) so the packaged migration and the runtime backend agree on one name.

litestar_queues.backends.sqlspec.schema.migration_directory() → Path[source]#: Return the packaged SQLSpec queue extension migration directory.

litestar_queues.backends.sqlspec.schema.migration_paths() → tuple[str, ...][source]#: Return packaged SQLSpec migration file paths.

litestar_queues.backends.sqlspec.schema.resolve_column_map(column_map: Mapping[str, str] | None = None) → dict[str, str][source]#: Return the default physical column map with adopter overrides applied.

litestar_queues.backends.sqlspec.schema.task_reservation_table_name_for(table_name: str) → str[source]#

Return the default forever-uniqueness reservation table for a queue table.

Schema-qualified names keep their schema and append DEFAULT_TASK_RESERVATION_TABLE_SUFFIX to the table part. Long derived names use the same portable deterministic shortening as maintenance tables.

litestar_queues.backends.sqlspec.schema.validate_column_map(column_map: Mapping[str, str]) → dict[str, str][source]#

Validate a canonical-to-adopter column map.

Returns:: A defensive copy of the validated map.
Raises:: QueueConfigurationError – If a canonical name is unknown or a mapped name is not a valid SQL identifier.

litestar_queues.backends.sqlspec.schema.validate_native_json_columns(columns: frozenset[str]) → frozenset[str][source]#

Validate native JSON passthrough columns.

Returns:: The validated column set.
Raises:: QueueConfigurationError – If any column is not a canonical JSON column.

litestar_queues.backends.sqlspec.schema.validate_table_name(table_name: str) → str[source]#

Validate a SQL identifier used for the queue table name.

Returns:: The validated table name, normalized to unquoted SQLSpec identifier parts.
Raises:: QueueConfigurationError – If the table name is not a valid SQL identifier.

SQLSpec queue store factory.

litestar_queues.backends.sqlspec.stores.factory.create_queue_store(config: SQLSpecStoreConfig, *, table_name: str | None = None, column_map: Mapping[str, str] | None = None, native_json_columns: frozenset[str] | None = None, manage_schema: bool = True) → SQLSpecQueueStore[source]#

Create a queue store for a SQLSpec adapter configuration.

Returns:: The queue store implementation for the SQLSpec adapter.

SQLSpec-backed queue event history.

class litestar_queues.backends.sqlspec.event_log.SQLSpecQueueEventLog(*, session_factory: Callable[[], AbstractAsyncContextManager[SQLSpecDriver]], datetime_serializer: Callable[[datetime], datetime | str], config: EventHistoryConfig, store: SQLSpecQueueEventLogStore, runtime_logger: logging.Logger | None = None)[source]#

Bases: object

Buffered SQLSpec event-history writer and query interface.

__init__(*, session_factory: Callable[[], AbstractAsyncContextManager[SQLSpecDriver]], datetime_serializer: Callable[[datetime], datetime | str], config: EventHistoryConfig, store: SQLSpecQueueEventLogStore, runtime_logger: logging.Logger | None = None) → None[source]#

property extra_columns: tuple[EventHistoryExtraColumn, ...]#: Declared extra scoping columns for this event log.

async publish_event(event: QueueEvent) → None[source]#: Buffer a queue event and flush when configured thresholds are reached.

async flush_events() → None[source]#: Flush buffered queue events through a SQLSpec session.

async query_events(query: QueueEventQuery | None = None, *, extra: Mapping[str, str] | None = None) → OffsetPagination[QueueEventLogRecord][source]#: Query durable event history records.

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary][source]#: Return per-stage event history aggregates.

async cleanup_events(*, before: datetime, match: QueueEventQuery | None = None, exclude: Sequence[QueueEventQuery] = (), limit: int | None = None) → int[source]#

Delete event history older than before.

limit bounds one batch, deleting the oldest matching rows first (oldest occurred_at, then event_id).

Returns:: Number of deleted event-history rows.

class litestar_queues.backends.sqlspec.event_log.SQLSpecQueueEventLogStore(*args: Any, extra_columns: Sequence[EventHistoryExtraColumn] | None = None, **kwargs: Any)[source]#

Bases: SQLSpecQueueStore

SQLSpec statement store for backend-managed queue event history.

__init__(*args: Any, extra_columns: Sequence[EventHistoryExtraColumn] | None = None, **kwargs: Any) → None[source]#

property extra_columns: tuple[EventHistoryExtraColumn, ...]#: Adopter-declared extra scoping columns on this event-history table.

create_statements() → list[str][source]#: Return statements that create the event-log table and indexes.

drop_statements() → list[str][source]#: Return statements that drop event-log artifacts.

insert_events_template() → str[source]#: Return a parametrized batch INSERT template for event rows.

parameter_name(column: str) → str[source]#: Return the bind parameter name for a public event column.

select_events(query: QueueEventQuery, extra: Mapping[str, str] | None = None) → Select[source]#

Return a SELECT for event-log records.

Raises:: ValueError – If extra names a column that was not declared.

summarize_stages(*, task_name: str | None = None) → tuple[str, dict[str, Any]][source]#: Return SQL and parameters for per-stage event summaries.

select_event_ids_before(*, before: DatetimeParam, limit: int | None) → Select[source]#

Return a SELECT of the oldest bounded event ids before a cutoff.

Raises:: ValueError – If limit is less than 1.

delete_events_by_ids(*, event_ids: Sequence[str]) → Delete[source]#: Return a DELETE statement scoped to the given event ids.

serialize_detail(detail: dict[str, Any]) → Any[source]#

Serialize event detail payloads with the SQLSpec JSON serializer.

Returns:: The adapter-shaped serialized detail payload.

deserialize_detail(value: Any) → dict[str, Any][source]#

Deserialize event detail payloads returned by a SQLSpec driver.

Returns:: The decoded detail mapping, or an empty mapping for non-object JSON.

class litestar_queues.backends.sqlspec.event_log.SpannerQueueEventLogStore(*args: Any, extra_columns: Sequence[EventHistoryExtraColumn] | None = None, **kwargs: Any)[source]#

Bases: SQLSpecQueueEventLogStore, SpannerQueueStore

Spanner event-log store using native DDL operations.

auto_native_json_columns: ClassVar[frozenset[str]] = frozenset({'detail'})#

create_statements() → list[str][source]#: Return Spanner-compatible event-log table and index statements.

drop_statements() → list[str][source]#: Return Spanner-compatible event-log DROP statements.

litestar_queues.backends.sqlspec.event_log.create_event_log_store(config: SQLSpecStoreConfig, *, queue_table_name: str, event_history_table_name: str | None = None, manage_schema: bool = True, extra_columns: Sequence[EventHistoryExtraColumn] = ()) → SQLSpecQueueEventLogStore[source]#

Create an event-log store for a SQLSpec adapter configuration.

Returns:: SQLSpec event-log store configured for the resolved event-log table.

litestar_queues.backends.sqlspec.event_log.resolve_event_history_table_name(queue_table_name: str, *, event_history_table_name: str | None = None) → str[source]#

Resolve the SQLSpec event-log table name for a queue table.

Returns:: The explicit event-log table name, or the derived queue-table event log name.

Advanced Alchemy#

Advanced Alchemy queue backend.

class litestar_queues.backends.advanced_alchemy.backend.SQLAlchemyBackend(config: QueueConfig | None = None, *, backend_config: SQLAlchemyBackendConfig | None = None)[source]#

Bases: BaseQueueBackend

SQLAlchemy queue backend using Advanced Alchemy services.

__init__(config: QueueConfig | None = None, *, backend_config: SQLAlchemyBackendConfig | None = None) → None[source]#: Initialize the queue backend.

property capabilities: QueueBackendCapabilities#: Backend behavior capabilities.

async open() → bool[source]#

Open Advanced Alchemy resources.

Returns:: True when resources are ready.

async close() → None[source]#: Close backend-owned resources.

get_event_log(config: EventHistoryConfig) → AdvancedAlchemyQueueEventLog[source]#: Return Advanced Alchemy-managed queue event history.

Persist a queued task.

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist multiple queued tasks in one Advanced Alchemy operation.

Returns:: Queue task records in input order.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#: Return a queued task by ID.

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#: Return a queued task by deduplication key.

async list_pending(*, limit: int = 1, queue: str | None = None, execution_backend: str | None = None) → list[QueuedTaskRecord][source]#: Return due pending or scheduled tasks ordered for execution.

async time_until_next_due(*, queues: tuple[str, ...] = ()) → float | None[source]#

Return seconds until the earliest not-yet-due pending/scheduled record.

Returns:: Seconds until the next due record, or None when there is no upcoming scheduled work.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#: Atomically claim a pending task.

Claim one task and report when this call expires that task.

Returns:: The claimed record and the expired record, at most one of which is set.

async claim_next(*, queues: tuple[str, ...] = (), execution_backend: str | None = None) → QueuedTaskRecord | None[source]#

Claim the next due task across the requested queues.

An empty queues tuple claims across all queues.

Returns:: The claimed task record, if one was available.

async claim_many(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → list[QueuedTaskRecord][source]#

Claim up to limit due tasks across the requested queues.

Returns:: Claimed task records.

async claim_many_with_expired(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#: Claim records and report expiry transitions from the same transactions.

async complete_task(task_id: UUID, *, result: Any = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Mark a task as completed.

Parameters:

task_id – Queue record identifier.
result – Task result payload.
expected_retry_count – When provided, update only if the record is still running with this retry count.

Mark a task as failed or retry it.

Parameters:

task_id – Queue record identifier.
error – Error message to persist.
retry – Whether retry policy may requeue the task.
expected_retry_count – When provided, update only if the record is still running with this retry count.
retry_at – Scheduled eligibility timestamp for a delayed retry.
queued_at – Queue ordering timestamp for the new retry attempt.

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist the owner of a running retry generation.

The write is fenced on status='running' AND retry_count = expected so a worker can never claim ownership of a generation it lost.

Parameters:

task_id – Queue record identifier.
worker_id – Identity to persist as the record owner.
expected_retry_count – Retry generation the caller believes it owns.

Raises:

NotImplementedError – Always; every backend must answer this.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending after interruption.

Parameters:

task_id – Queue record identifier.
expected_retry_count – Retry generation the caller owns.
worker_id – Identity that must currently own the record.
queued_at – Requeue timestamp used for fair claim ordering.

Raises:

NotImplementedError – Always; every backend must answer this.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

Cancel a task.

Parameters:

task_id – ID of the task to cancel.
include_running – Whether to cancel a task that is currently running.
expected_retry_count – Only cancel if the task is still on this retry generation.

Returns:

True if the task was successfully cancelled, False otherwise.

Cancel tasks matching a domain predicate.

Parameters:

task_name – Optional task name exact match.
queue – Optional queue exact match.
kwargs – Optional top-level kwargs exact-match subset.
metadata – Optional top-level metadata exact-match subset.
include_running – When true, running records are included for cooperative cancellation.

Returns:

Number of records cancelled.

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

Update heartbeat timestamps for running tasks.

Returns:: The task IDs confirmed touched or missed by the backend.
Raises:: NotImplementedError – Always; every backend must answer this.

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#

Clear heartbeat timestamps for task IDs.

Parameters:

task_ids – Queue record identifiers.
expected_retry_count – When provided, clear only records that still match this retry count.

Raises:

NotImplementedError – Always; every backend must answer this.

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None) → StaleTaskRecoveryResult[source]#

Recover running tasks with stale heartbeats.

Parameters:

stale_after – Heartbeat age past which a running task is stale.
limit – When provided, recover at most this many records ordered oldest-first (oldest heartbeat, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

Summary of requeued, failed, skipped, and handler-needed records.

Raises:

NotImplementedError – Always; every backend must answer this.

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an external execution reference for a running task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

Atomically reserve a due, unexpired task for external dispatch.

The default rejects dispatch because a read-then-write fallback cannot protect the external side effect.

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#: Clear an exact pending external attempt and wake dispatchers.

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#: Atomically rotate an exact pending external attempt reference.

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Release a matching external-dispatch reservation.

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Replace an owned dispatch reservation with its execution reference.

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an execution backend/profile change for a queued task.

Returns:: The updated queued task record, if one exists.
Raises:: NotImplementedError – Always; every backend must answer this.

async notify_new_task(record: QueuedTaskRecord) → None[source]#: Publish a PostgreSQL worker wakeup marker when enabled.

async notify_new_tasks(records: Sequence[QueuedTaskRecord]) → None[source]#: Coalesce a batch of task records into at most one wakeup marker.

async wait_for_wakeups(timeout: float | None = None) → bool[source]#

Wait for a PostgreSQL worker wakeup marker when configured.

Returns:: True when a wakeup marker or due-row reconciliation is observed.

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Return externally dispatched tasks with references to reconcile.

Raises:: NotImplementedError – Always; every backend must answer this.

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#

Return queue status counts.

Raises:: NotImplementedError – Always; every backend must answer this.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

Transition overdue pending or scheduled records to expired.

Returns:: Records transitioned to expired.
Raises:: NotImplementedError – Always; every backend must answer this.

async list_completed_by_task(task_name: str, *, since: datetime | None = None, limit: int = 10) → list[QueuedTaskRecord][source]#

Return recent completed records for a task name.

Raises:: NotImplementedError – Always; every backend must answer this.

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

Delete terminal records completed before a cutoff.

Routine terminal cleanup never touches unique_until="forever" reservations; only reset_identity() removes them.

Parameters:

before – Delete terminal records completed strictly before this UTC cutoff.
limit – When provided, delete at most this many records ordered oldest-first (oldest completed_at, then record id). None preserves the historical unbounded behavior; bounded maintenance always supplies a positive limit.

Returns:

The number of deleted records.

Raises:

NotImplementedError – Always; every backend must answer this.

async acquire_maintenance(name: str, token: str, *, ttl: timedelta) → bool[source]#

Acquire maintenance ownership via a portable compare-and-set.

Updates an existing expired row for name to this token; if no row was expired, inserts a fresh one inside a savepoint so a uniqueness race (live ownership held elsewhere) is treated as a denied acquisition.

Returns:: True when maintenance ownership is held under token.

async release_maintenance(name: str, token: str) → bool[source]#

Release maintenance ownership only when token matches the holder.

Returns:: True when ownership held under token was deleted.

async reserve_identity(key: str, *, task_id: UUID, task_name: str) → TaskReservation | None[source]#

Reserve a forever identity via select-then-insert with an integrity fallback.

The reservation table’s unique identity_key column is the atomicity arbiter: a losing concurrent insert surfaces an integrity error and the loser re-reads the winning owner. The reservation table is separate from the task table and terminal cleanup never touches it.

Returns:: None when this caller won the reservation; otherwise the existing owner reservation.

async has_identity(key: str) → TaskReservation | None[source]#: Return the reservation owning a reserved forever identity, if any.

async reset_identity(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete a forever identity reservation via atomic compare-and-delete.

Parameters:

key – The exact effective identity key.
expected_task_id – Optional task owner required for deletion.

Returns:

True when a reservation was removed.

Advanced Alchemy backend configuration.

class litestar_queues.backends.advanced_alchemy.config.SQLAlchemyBackendConfig(sqlalchemy_config: SQLAlchemyAsyncConfig | None = None, heartbeat_session_maker: async_sessionmaker[AsyncSession] | None = None, model_class: type[object] | None = <factory>, event_history_model_class: type[object] | None = <factory>, maintenance_model_class: type[object] | None = <factory>, task_reservation_model_class: type[object] | None = <factory>, worker_wakeups: bool = False, wakeup_channel: str | None = None, wakeup_poll_interval: float | None = None)[source]#

Bases: object

Configuration values for the SQLAlchemy queue backend.

backend_name: ClassVar[str] = 'advanced-alchemy'#

sqlalchemy_config: SQLAlchemyAsyncConfig | None#: Advanced Alchemy async configuration; None requires an explicit session maker.

heartbeat_session_maker: async_sessionmaker[AsyncSession] | None#: Dedicated heartbeat session factory; None reuses the configured database path.

model_class: type[object] | None#: Queue-task ORM model; None disables package-managed task persistence.

event_history_model_class: type[object] | None#: Task-event history ORM model; None disables history support.

maintenance_model_class: type[object] | None#: Maintenance coordination ORM model; None disables maintenance support.

__init__(sqlalchemy_config: SQLAlchemyAsyncConfig | None = None, heartbeat_session_maker: async_sessionmaker[AsyncSession] | None = None, model_class: type[object] | None = <factory>, event_history_model_class: type[object] | None = <factory>, maintenance_model_class: type[object] | None = <factory>, task_reservation_model_class: type[object] | None = <factory>, worker_wakeups: bool = False, wakeup_channel: str | None = None, wakeup_poll_interval: float | None = None) → None#

task_reservation_model_class: type[object] | None#: Permanent task-reservation ORM model; None disables durable reservations.

worker_wakeups: bool#: Whether workers listen for database wakeup hints between polling passes.

wakeup_channel: str | None#: Database notification channel; None derives it from QueueConfig.namespace.

wakeup_poll_interval: float | None#: Wakeup-listener fallback poll interval in seconds; None uses the backend default.

Advanced Alchemy queue task mixins.

class litestar_queues.backends.advanced_alchemy.mixins.QueueEventHistoryModelMixin[source]#

Bases: object

Declarative mixin carrying generic queue event-history columns and indexes.

event_id = <sqlalchemy.orm.properties.MappedColumn object>#

event_type = <sqlalchemy.orm.properties.MappedColumn object>#

task_id = <sqlalchemy.orm.properties.MappedColumn object>#

task_name = <sqlalchemy.orm.properties.MappedColumn object>#

queue = <sqlalchemy.orm.properties.MappedColumn object>#

worker_id = <sqlalchemy.orm.properties.MappedColumn object>#

execution_backend = <sqlalchemy.orm.properties.MappedColumn object>#

execution_profile = <sqlalchemy.orm.properties.MappedColumn object>#

actor_type = <sqlalchemy.orm.properties.MappedColumn object>#

actor_id = <sqlalchemy.orm.properties.MappedColumn object>#

level = <sqlalchemy.orm.properties.MappedColumn object>#

message = <sqlalchemy.orm.properties.MappedColumn object>#

stage = <sqlalchemy.orm.properties.MappedColumn object>#

scope = <sqlalchemy.orm.properties.MappedColumn object>#

scope_key = <sqlalchemy.orm.properties.MappedColumn object>#

actor = <sqlalchemy.orm.properties.MappedColumn object>#

entity = <sqlalchemy.orm.properties.MappedColumn object>#

detail_json = <sqlalchemy.orm.properties.MappedColumn object>#

progress_current = <sqlalchemy.orm.properties.MappedColumn object>#

progress_total = <sqlalchemy.orm.properties.MappedColumn object>#

progress_percent = <sqlalchemy.orm.properties.MappedColumn object>#

duration_ms = <sqlalchemy.orm.properties.MappedColumn object>#

sequence = <sqlalchemy.orm.properties.MappedColumn object>#

occurred_at = <sqlalchemy.orm.properties.MappedColumn object>#

class litestar_queues.backends.advanced_alchemy.mixins.QueueMaintenanceModelMixin[source]#

Bases: object

Declarative mixin carrying distributed maintenance coordination columns.

Compose this with an application-owned Advanced Alchemy base that provides a compatible id primary key. Adopter-owned model and migration setups must include the resulting table; the queue backend never calls metadata.create_all.

name = <sqlalchemy.orm.properties.MappedColumn object>#

token = <sqlalchemy.orm.properties.MappedColumn object>#

expires_at = <sqlalchemy.orm.properties.MappedColumn object>#

class litestar_queues.backends.advanced_alchemy.mixins.QueueTaskModelMixin[source]#

Bases: object

Declarative mixin carrying queue task columns and indexes.

Compose this with an application-owned Advanced Alchemy base that provides compatible id and created_at columns.

task_name = <sqlalchemy.orm.properties.MappedColumn object>#

args_json = <sqlalchemy.orm.properties.MappedColumn object>#

kwargs_json = <sqlalchemy.orm.properties.MappedColumn object>#

queue = <sqlalchemy.orm.properties.MappedColumn object>#

execution_backend = <sqlalchemy.orm.properties.MappedColumn object>#

execution_profile = <sqlalchemy.orm.properties.MappedColumn object>#

execution_ref = <sqlalchemy.orm.properties.MappedColumn object>#

worker_id = <sqlalchemy.orm.properties.MappedColumn object>#

status = <sqlalchemy.orm.properties.MappedColumn object>#

priority = <sqlalchemy.orm.properties.MappedColumn object>#

max_retries = <sqlalchemy.orm.properties.MappedColumn object>#

retry_count = <sqlalchemy.orm.properties.MappedColumn object>#

scheduled_at = <sqlalchemy.orm.properties.MappedColumn object>#

expires_at = <sqlalchemy.orm.properties.MappedColumn object>#

queued_at = <sqlalchemy.orm.properties.MappedColumn object>#

started_at = <sqlalchemy.orm.properties.MappedColumn object>#

completed_at = <sqlalchemy.orm.properties.MappedColumn object>#

heartbeat_at = <sqlalchemy.orm.properties.MappedColumn object>#

result_json = <sqlalchemy.orm.properties.MappedColumn object>#

error = <sqlalchemy.orm.properties.MappedColumn object>#

task_key = <sqlalchemy.orm.properties.MappedColumn object>#

metadata_json = <sqlalchemy.orm.properties.MappedColumn object>#

class litestar_queues.backends.advanced_alchemy.mixins.QueueTaskReservationModelMixin[source]#

Bases: object

Declarative mixin carrying forever-uniqueness reservation columns.

Compose this with an application-owned Advanced Alchemy base that provides compatible id and created_at columns. The unique identity_key column is the atomicity arbiter for forever reservations; the table is separate from the queue task table so routine terminal cleanup never touches it. It stores only the identity key and the originating task id/name.

identity_key = <sqlalchemy.orm.properties.MappedColumn object>#

task_id = <sqlalchemy.orm.properties.MappedColumn object>#

task_name = <sqlalchemy.orm.properties.MappedColumn object>#

Advanced Alchemy queue task repository.

class litestar_queues.backends.advanced_alchemy.repository.QueueEventLogRepository(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, cache_manager: CacheManager | None = None, bind_group: str | None = None, **kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepository[Any]

Repository for queue event-history records.

classmethod for_model(model_class: type[QueueEventHistoryModelMixin]) → type['QueueEventLogRepository'][source]#: Return a repository subclass bound to model_class.

class litestar_queues.backends.advanced_alchemy.repository.QueueTaskRepository(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, cache_manager: CacheManager | None = None, bind_group: str | None = None, **kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepository[Any]

Repository for queue task records.

classmethod for_model(model_class: type[QueueTaskModelMixin]) → type['QueueTaskRepository'][source]#: Return a repository subclass bound to model_class.

class litestar_queues.backends.advanced_alchemy.repository.QueueTaskReservationRepository(*, statement: ~sqlalchemy.sql.selectable.Select | None = None, session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, wrap_exceptions: bool = True, uniquify: bool | None = None, count_with_window_function: bool | None = None, cache_manager: CacheManager | None = None, bind_group: str | None = None, **kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepository[Any]

Repository for forever-uniqueness reservations.

classmethod for_model(model_class: type[QueueTaskReservationModelMixin]) → type['QueueTaskReservationRepository'][source]#: Return a repository subclass bound to model_class.

Advanced Alchemy queue persistence service.

class litestar_queues.backends.advanced_alchemy.service.QueueEventLogService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[Any]

Persistence operations for Advanced Alchemy queue event-history records.

classmethod for_model(model_class: type[QueueEventHistoryModelMixin]) → type['QueueEventLogService'][source]#: Return a service subclass bound to model_class.

async add_records(records: Sequence[QueueEventLogRecord]) → None[source]#: Persist event-history records.

async query_events(query: QueueEventQuery) → tuple[int, list[QueueEventLogRecord]][source]#

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary][source]#

async cleanup_events(before: datetime, *, limit: int | None = None, match: QueueEventQuery | None = None, exclude: tuple[QueueEventQuery, ...] | None = None) → int[source]#

Delete event-history records older than before.

limit bounds one batch, deleting the oldest matching rows first (oldest occurred_at, then id).

Returns:: Number of deleted event-history rows.

model_from_record(record: QueueEventLogRecord) → Any[source]#

Convert a backend-neutral event-history record into an ORM model.

Returns:: Advanced Alchemy event-history model.

static record_from_model(model: Any) → QueueEventLogRecord[source]#

Convert an ORM model into a backend-neutral event-history record.

Returns:: Backend-neutral event-history record.

class litestar_queues.backends.advanced_alchemy.service.QueueTaskReservationService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[Any]

Persistence operations for forever-uniqueness reservations.

classmethod for_model(model_class: type[QueueTaskReservationModelMixin]) → type['QueueTaskReservationService'][source]#: Return a service subclass bound to model_class.

async reserve(key: str, *, task_id: UUID, task_name: str) → Any | None[source]#

Reserve key by select-then-insert within the caller’s transaction.

Returns:: None when the reservation was inserted; otherwise the existing owner model.

async get_owner(key: str) → Any | None[source]#: Return the reservation model owning key, if any.

async delete_by_key(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete the reservation for key with optional owner fencing.

Parameters:

key – The exact effective identity key.
expected_task_id – Optional task owner required for deletion.

Returns:

True when a reservation row was removed.

class litestar_queues.backends.advanced_alchemy.service.QueueTaskService(session: ~sqlalchemy.ext.asyncio.session.AsyncSession | ~sqlalchemy.ext.asyncio.scoping.async_scoped_session[~sqlalchemy.ext.asyncio.session.AsyncSession], *, statement: ~sqlalchemy.sql.selectable.Select | None = None, auto_expunge: bool = False, auto_refresh: bool = True, auto_commit: bool = False, order_by: ~typing.List[tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any]] | tuple[str | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any], bool] | ~sqlalchemy.sql.elements.UnaryExpression[~typing.Any] | None = None, error_messages: ~advanced_alchemy.exceptions.ErrorMessages | type[~advanced_alchemy.utils.dataclass.Empty] | None = <class 'advanced_alchemy.utils.dataclass.Empty'>, wrap_exceptions: bool = True, load: ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~collections.abc.Sequence[~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any]]] | ~sqlalchemy.orm.strategy_options._AbstractLoad | ~typing.Literal['*'] | ~sqlalchemy.orm.attributes.InstrumentedAttribute[~typing.Any] | ~sqlalchemy.orm.relationships.RelationshipProperty[~typing.Any] | ~sqlalchemy.orm.interfaces.MapperProperty[~typing.Any] | ~sqlalchemy.sql.base.ExecutableOption | ~collections.abc.Sequence[~sqlalchemy.sql.base.ExecutableOption] | None = None, execution_options: dict[str, ~typing.Any] | None = None, uniquify: bool | None = None, count_with_window_function: bool | None = None, **repo_kwargs: ~typing.Any)[source]#

Bases: SQLAlchemyAsyncRepositoryService[Any]

Persistence operations for Advanced Alchemy queue records.

classmethod for_model(model_class: type[QueueTaskModelMixin]) → type['QueueTaskService'][source]#: Return a service subclass bound to model_class.

async enqueue(task_name: str, *, args: tuple[Any, ...], kwargs: dict[str, Any], queue: str, priority: int, max_retries: int, scheduled_at: datetime | None, expires_at: datetime | None = None, key: str | None, execution_backend: str, execution_profile: str | None, metadata: dict[str, Any], id: UUID | None = None) → QueuedTaskRecord[source]#

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist many task requests in the current repository transaction.

Returns:: Queue task records in input order.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#

async list_pending(*, limit: int, queue: str | None, execution_backend: str | None) → list[QueuedTaskRecord][source]#

async next_scheduled_at(*, queues: Sequence[str] = ()) → datetime | None[source]#

Return the earliest not-yet-due scheduled_at among pending/scheduled records.

Used to bound the worker’s adaptive polling wait so a scheduled or retried task is never discovered later than its own due time.

Returns:: The earliest future scheduled_at, or None when there is no upcoming scheduled work.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#

async claim_task_with_expired(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → tuple[QueuedTaskRecord | None, QueuedTaskRecord | None][source]#: Claim one task and identify a claim-time expiry owned by this call.

async claim_next(*, queue: str | None, execution_backend: str | None) → QueuedTaskRecord | None[source]#

async claim_many(*, limit: int, queue: str | None, execution_backend: str | None) → list[QueuedTaskRecord][source]#

Claim up to limit records in the current repository transaction.

Returns:: Claimed task records.

async claim_many_with_expired(*, limit: int, queue: str | None, execution_backend: str | None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#: Claim records and report all expiry transitions owned by this transaction.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

async complete_task(task_id: UUID, *, result: Any | None = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

async fail_task(task_id: UUID, error: str, *, retry: bool, expected_retry_count: int | None = None, retry_at: datetime | None = None, queued_at: datetime | None = None) → QueuedTaskRecord | None[source]#

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist running-record ownership behind a status/generation fence.

Returns:: The owned record, or None when the fence was lost.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending behind an owner/generation fence.

Returns:: The requeued record, or None when the fence was lost.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

async cancel_tasks(*, task_name: str | None = None, queue: str | None = None, kwargs: Mapping[str, Any] | None = None, metadata: Mapping[str, Any] | None = None, include_running: bool = False) → int[source]#

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None, priority_policy: StaleRequeuePriority = 'preserve') → StaleTaskRecoveryResult[source]#

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None) → QueuedTaskRecord | None[source]#

async reserve_external_dispatch(task_id: UUID, execution_backend: str, reservation_ref: str, *, execution_profile: str | None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None) → QueuedTaskRecord | None[source]#

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None) → QueuedTaskRecord | None[source]#

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None) → QueuedTaskRecord | None[source]#

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#

async list_completed_by_task(task_name: str, *, since: datetime | None, limit: int) → list[QueuedTaskRecord][source]#

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

model_from_record(record: QueuedTaskRecord) → Any[source]#

Convert a backend-neutral record into an Advanced Alchemy model.

Returns:: The Advanced Alchemy queue task model.

static record_from_model(model: Any) → QueuedTaskRecord[source]#

Convert an Advanced Alchemy model into a backend-neutral record.

Returns:: The backend-neutral queued task record.

Advanced Alchemy-backed queue event history.

class litestar_queues.backends.advanced_alchemy.event_log.AdvancedAlchemyQueueEventLog(config: EventHistoryConfig, *, service_factory: Callable[[], AbstractAsyncContextManager['QueueEventLogService']], transaction_factory: Callable[[], AbstractAsyncContextManager['QueueEventLogService']], runtime_logger: logging.Logger | None = None)[source]#

Bases: object

Buffered Advanced Alchemy event-history writer and query interface.

__init__(config: EventHistoryConfig, *, service_factory: Callable[[], AbstractAsyncContextManager['QueueEventLogService']], transaction_factory: Callable[[], AbstractAsyncContextManager['QueueEventLogService']], runtime_logger: logging.Logger | None = None) → None[source]#

async publish_event(event: QueueEvent) → None[source]#: Buffer a queue event and flush when configured thresholds are reached.

async flush_events() → None[source]#: Flush buffered queue events through an Advanced Alchemy session.

async query_events(query: QueueEventQuery | None = None, *, extra: Mapping[str, str] | None = None) → OffsetPagination[QueueEventLogRecord][source]#: Query durable event history records.

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary][source]#: Return per-stage event history aggregates.

async cleanup_events(*, before: datetime, match: QueueEventQuery | None = None, exclude: Sequence[QueueEventQuery] = (), limit: int | None = None) → int[source]#

Delete event history older than before.

Returns:: Number of deleted event-history rows.

Advanced Alchemy queue models.

class litestar_queues.backends.advanced_alchemy.models.QueueEventHistoryModel(**kwargs)[source]#

Bases: UUIDAuditBase, QueueEventHistoryModelMixin

Default queue event-history model for the Advanced Alchemy backend.

__init__(**kwargs)#

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

actor#

actor_id#

actor_type#

created_at: Mapped[datetime.datetime]#: Date/time of instance creation.

detail_json#

duration_ms#

entity#

event_id#

event_type#

execution_backend#

execution_profile#

id: Mapped[UUID]#: UUID Primary key column.

level#

message#

occurred_at#

progress_current#

progress_percent#

progress_total#

queue#

scope#

scope_key#

sequence#

stage#

task_id#

task_name#

updated_at: Mapped[datetime.datetime]#: Date/time of instance last update.

worker_id#

class litestar_queues.backends.advanced_alchemy.models.QueueMaintenanceModel(**kwargs)[source]#

Bases: UUIDAuditBase, QueueMaintenanceModelMixin

Default distributed maintenance coordination model for the Advanced Alchemy backend.

__init__(**kwargs)#

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

created_at: Mapped[datetime.datetime]#: Date/time of instance creation.

expires_at#

id: Mapped[UUID]#: UUID Primary key column.

name#

token#

updated_at: Mapped[datetime.datetime]#: Date/time of instance last update.

class litestar_queues.backends.advanced_alchemy.models.QueueTaskModel(**kwargs)[source]#

Bases: UUIDAuditBase, QueueTaskModelMixin

Default queue task model for the Advanced Alchemy backend.

__init__(**kwargs)#

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

args_json#

completed_at#

created_at: Mapped[datetime.datetime]#: Date/time of instance creation.

error#

execution_backend#

execution_profile#

execution_ref#

expires_at#

heartbeat_at#

id: Mapped[UUID]#: UUID Primary key column.

kwargs_json#

max_retries#

metadata_json#

priority#

queue#

queued_at#

result_json#

retry_count#

scheduled_at#

started_at#

status#

task_key#

task_name#

updated_at: Mapped[datetime.datetime]#: Date/time of instance last update.

worker_id#

class litestar_queues.backends.advanced_alchemy.models.QueueTaskReservationModel(**kwargs)[source]#

Bases: UUIDAuditBase, QueueTaskReservationModelMixin

Default forever-uniqueness reservation model for the Advanced Alchemy backend.

__init__(**kwargs)#

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

created_at: Mapped[datetime.datetime]#: Date/time of instance creation.

id: Mapped[UUID]#: UUID Primary key column.

identity_key#

task_id#

task_name#

updated_at: Mapped[datetime.datetime]#: Date/time of instance last update.

Redis#

Redis queue backend.

Stores queued task records in a Redis-protocol key-value server. The implementation lives directly on RedisQueueBackend; the Valkey backend inherits from this class and only swaps the client factory and _backend_name ClassVar.

class litestar_queues.backends.redis.backend.RedisQueueBackend(config: QueueConfig | None = None, *, backend_config: _RedisBackendConfig | None = None)[source]#

Bases: BaseQueueBackend

Queue backend that stores records in a Redis-protocol key-value server.

Ready work lives in one global {prefix}:ready sorted set scored priority-major / created_at-minor, so the claim pops the globally-correct next task with one ordered ZRANGE instead of a Python-side sort over an HGETALL of every due candidate. A separate {prefix}:scheduled sorted set scored by scheduled_at preserves exact delayed-promotion due-gating: the claim script promotes now-due scheduled ids into ready before scanning, so future-scheduled tasks are never claimable early. Keeping one global ready set rather than per-queue sets makes the claim a single EVAL with no queue enumeration; the queue and execution_backend filters skip non-matching top entries inside the script.

Ready scores are IEEE-754 doubles, exact for integers up to 2^53. With stride 1e13 and created_ms near 1.7e12 the priority band (-priority) * 1e13 stays exact for abs(priority) <= 450, far beyond realistic priorities; ties break on created_ms ascending at millisecond resolution.

All Lua scripts build their keys from a key_prefix ARG via string concatenation, which is single-node/replica only. Redis Cluster is out of scope: multi-key scripts on a cluster require same-slot hash-tagged keys and no hash-tag support is added. The composite-score ready/scheduled layout replaces the old {prefix}:pending zset outright with no data migration; records enqueued under the old layout are stranded (benchmark namespaces are ephemeral).

__init__(config: QueueConfig | None = None, *, backend_config: _RedisBackendConfig | None = None) → None[source]#: Initialize the queue backend.

property capabilities: QueueBackendCapabilities#: Backend behavior capabilities.

async open() → bool[source]#

Open Redis-protocol client resources.

Returns:: True when the client is ready.

async close() → None[source]#: Close owned Redis-protocol client resources.

get_event_log(config: EventHistoryConfig) → RedisQueueEventLog[source]#: Return a backend-owned queue event history implementation, if supported.

Persist a queued task.

Returns:: The created or deduplicated queued task record.

async enqueue_many(requests: Sequence[TaskRequest]) → list[QueuedTaskRecord][source]#

Persist a batch of Redis-backed tasks and coalesce worker wakeups.

Returns:: Queue task records in the same order as requests.

async get_task(task_id: UUID) → QueuedTaskRecord | None[source]#: Return a queued task by ID.

async get_task_by_key(key: str) → QueuedTaskRecord | None[source]#: Return a queued task by deduplication key.

async list_pending(*, limit: int = 1, queue: str | None = None, execution_backend: str | None = None) → list[QueuedTaskRecord][source]#: Return due pending or scheduled tasks ordered for execution.

async claim_task(task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → QueuedTaskRecord | None[source]#: Atomically claim a pending task.

Atomically claim a pending task via a single fenced script.

Returns:: The claimed record and the expired record, at most one of which is set.

async claim_many(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → list[QueuedTaskRecord][source]#

Claim up to limit due tasks in a single fenced EVAL.

Returns:: Claimed task records in claim order.

async claim_many_with_expired(*, limit: int, queues: tuple[str, ...] = (), execution_backend: str | None = None, queue_limits: Mapping[str, int] | None = None) → tuple[list[QueuedTaskRecord], list[QueuedTaskRecord]][source]#: Claim records and report expirations owned by the same Lua script.

async complete_task(task_id: UUID, *, result: Any | None = None, expected_retry_count: int | None = None) → QueuedTaskRecord | None[source]#

Mark a task as completed via a single fenced script.

Returns:: The completed record, if it exists.

Mark a task as failed or retry it via a single fenced script.

Returns:: The updated record, if it exists.

async assign_worker(task_id: UUID, *, worker_id: str, expected_retry_count: int) → QueuedTaskRecord | None[source]#

Persist running-record ownership through the fenced transition script.

Returns:: The owned record, or None when the fence was lost.

async interrupt_task(task_id: UUID, *, expected_retry_count: int, worker_id: str, queued_at: datetime) → QueuedTaskRecord | None[source]#

Return an owned running attempt to pending through the fenced transition script.

Returns:: The requeued record, or None when the fence was lost.

async cancel_task(task_id: UUID, *, include_running: bool = False, expected_retry_count: int | None = None) → bool[source]#

Cancel a task via a single fenced script.

Returns:: True when the task was cancelled.

Cancel tasks matching a domain predicate.

Returns:: Number of records cancelled.

async touch_heartbeats(touches: Sequence[HeartbeatTouch]) → HeartbeatTouchResult[source]#

Update heartbeat timestamps for running tasks.

Returns:: The task IDs confirmed touched or missed by the backend.

async null_heartbeats(task_ids: list[UUID], *, expected_retry_count: int | None = None) → None[source]#: Clear heartbeat timestamps for task IDs via a fenced script.

async requeue_stale_running(*, stale_after: timedelta, limit: int | None = None) → StaleTaskRecoveryResult[source]#

Requeue running tasks with stale heartbeats.

Candidates are ordered oldest-heartbeat-first (then by id) and capped at limit before any mutation so one maintenance batch is bounded.

Returns:: Summary of recovered records.

async set_execution_ref(task_id: UUID, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an external execution reference for a running task via a fenced script.

Returns:: The updated record, if it exists.

Atomically reserve a due, unexpired task for external dispatch.

The default rejects dispatch because a read-then-write fallback cannot protect the external side effect.

async clear_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str) → QueuedTaskRecord | None[source]#: Clear an exact pending external attempt and wake dispatchers.

async replace_execution_ref(task_id: UUID, expected_retry_count: int, expected_execution_ref: str, execution_ref: str) → QueuedTaskRecord | None[source]#: Atomically rotate an exact pending external attempt reference.

async release_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Release a matching external-dispatch reservation.

async finalize_external_dispatch(task_id: UUID, reservation_ref: str, execution_backend: str, execution_ref: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#: Replace an owned dispatch reservation with its execution reference.

async set_execution_backend(task_id: UUID, execution_backend: str, *, execution_profile: str | None = None) → QueuedTaskRecord | None[source]#

Persist an execution backend/profile change for a queued task via a fenced script.

Returns:: The updated record, if it exists.

async list_running_external(*, limit: int | None = None) → list[QueuedTaskRecord][source]#: Return externally dispatched tasks with references to reconcile.

async get_statistics(*, queue: str | None = None) → QueueStatistics[source]#: Return queue status counts.

async expire_overdue(*, limit: int | None = None) → list[QueuedTaskRecord][source]#: Transition overdue pending and scheduled records to expired.

async list_completed_by_task(task_name: str, *, since: datetime | None = None, limit: int = 10) → list[QueuedTaskRecord][source]#: Return recent completed records for a task name.

async cleanup_terminal(before: datetime, *, limit: int | None = None) → int[source]#

Delete terminal records completed before a cutoff.

Candidates are ordered oldest-completion-first (then by id) and capped at limit before any deletion so one maintenance batch is bounded.

Returns:: Number of deleted records.

async rebuild_maintenance_indexes() → int[source]#

Rebuild ordered maintenance indexes for a populated pre-index namespace.

This is an intentionally unbounded, one-time upgrade operation. Stop all queue writers using this Redis/Valkey namespace before calling it. Interrupted calls are safe to retry because the version marker is written only after every task has been reindexed.

Returns:: Number of queue records indexed.

async acquire_maintenance(name: str, token: str, *, ttl: timedelta) → bool[source]#

Acquire namespaced SET NX PX maintenance ownership.

Returns:: True when ownership was set for token.

async release_maintenance(name: str, token: str) → bool[source]#

Release maintenance ownership via token-checked Lua compare-and-delete.

Returns:: True when ownership held under token was deleted.

async reserve_identity(key: str, *, task_id: UUID, task_name: str) → TaskReservation | None[source]#

Reserve a forever identity via an atomic HGET-or-HSET script.

The task-reservation hash is separate from :task:/:keys and is never touched by terminal cleanup.

Returns:: None when this caller won the reservation; otherwise the existing owner reservation.

async has_identity(key: str) → TaskReservation | None[source]#: Return the reservation owning a reserved forever identity, if any.

async reset_identity(key: str, *, expected_task_id: UUID | None = None) → bool[source]#

Delete a forever identity reservation via atomic compare-and-delete.

Parameters:

key – The exact effective identity key.
expected_task_id – Optional task owner required for deletion.

Returns:

True when a reservation was removed.

async notify_new_task(record: QueuedTaskRecord) → None[source]#: Publish a Redis-protocol pub/sub message when work is available.

async wait_for_wakeups(timeout: float | None = None) → bool[source]#

Wait for a Redis-protocol pub/sub message when notifications are enabled.

A single pub/sub receive is retained across worker poll timeouts; only a real message, a read failure, or backend close ends it.

Returns:: True when a notification was observed.

async notify_worker_control(worker_id: str | None) → None[source]#

Publish a worker-control hint on the Redis-protocol control channel.

The hint is lossy by contract: it only shortens the wait before the owning worker reconciles durable status.

async wait_for_worker_control(*, worker_id: str, timeout: float | None = None) → bool[source]#

Wait for a Redis-protocol worker-control hint.

The control subscription and its pending receive are retained across worker poll timeouts, exactly like the wakeup subscription, and are never shared with it: one in-flight read per subscription.

Returns:: True when a control hint was observed.
Raises:: Exception – Whatever the pub/sub receive raised, after the subscription is reset so the next wait reconnects.

async time_until_next_due(*, queues: tuple[str, ...] = ()) → float | None[source]#

Return seconds until the earliest not-yet-due scheduled record.

Reads the lowest-scored member of the global scheduled sorted set (scored by scheduled_at): an O(1) lookup independent of queue size. queues is not applied because the sorted set is not queue-scoped; an unfiltered bound is always safe here (it can only wake the worker sooner than strictly necessary, never later).

Returns:: Seconds until the next due record, or None when there is no upcoming scheduled work.

async wait_for_completion(task_id: UUID, *, timeout: float | None = None) → bool[source]#

Wait for a terminal completion message naming task_id.

Returns:: True when a completion signal for task_id arrived before the deadline.

Redis queue backend configuration.

class litestar_queues.backends.redis.config.RedisBackendConfig(url: str = 'redis://localhost:6379/0', key_prefix: str | None = None, worker_wakeups: bool = True, wakeup_channel: str | None = None, client: Redis | None = None)[source]#

Bases: object

Configuration for the Redis queue backend.

backend_name: ClassVar[str] = 'redis'#

url: str#: Redis connection URL used when no client is injected.

key_prefix: str | None#: Prefix applied to every queue key; None derives it from QueueConfig.namespace.

worker_wakeups: bool#: Whether workers subscribe for Redis wakeup hints between polling passes.

__init__(url: str = 'redis://localhost:6379/0', key_prefix: str | None = None, worker_wakeups: bool = True, wakeup_channel: str | None = None, client: Redis | None = None) → None#

wakeup_channel: str | None#: Worker-wakeup pub/sub channel; None derives it from QueueConfig.namespace.

client: Redis | None#: Injected async Redis client; None creates one from url.

Redis-protocol queue event history.

class litestar_queues.backends.redis.event_log.RedisQueueEventLog(*, backend: RedisQueueBackend, config: EventHistoryConfig)[source]#

Bases: object

Buffered Redis-protocol event-history writer and query interface.

__init__(*, backend: RedisQueueBackend, config: EventHistoryConfig) → None[source]#

async publish_event(event: QueueEvent) → None[source]#: Buffer a queue event and flush when configured thresholds are reached.

async flush_events() → None[source]#: Flush buffered queue events through a Redis pipeline.

async query_events(query: QueueEventQuery | None = None, *, extra: Mapping[str, str] | None = None) → OffsetPagination[QueueEventLogRecord][source]#

Return a filtered, ordered page of event history records.

Returns:: The matching page.

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary][source]#: Return per-stage event history aggregates.

async cleanup_events(*, before: datetime, match: QueueEventQuery | None = None, exclude: Sequence[QueueEventQuery] = (), limit: int | None = None) → int[source]#

Delete event history older than before.

Returns:: Number of removed event-history records.

Valkey#

Valkey queue backend.

The Valkey wire protocol is API-compatible with Redis, so this backend inherits the full RedisQueueBackend implementation and only overrides the client factory (uses valkey.asyncio instead of redis.asyncio) plus the _backend_name ClassVar that drives lock-name error messages and the valkey-pubsub notification capability label.

class litestar_queues.backends.valkey.backend.ValkeyQueueBackend(config: QueueConfig | None = None, *, backend_config: ValkeyBackendConfig | None = None)[source]#

Bases: RedisQueueBackend

Valkey-backed queue backend.

__init__(config: QueueConfig | None = None, *, backend_config: ValkeyBackendConfig | None = None) → None[source]#: Initialize the queue backend.

Valkey queue backend configuration.

class litestar_queues.backends.valkey.config.ValkeyBackendConfig(url: str = 'redis://localhost:6379/0', key_prefix: str | None = None, worker_wakeups: bool = True, wakeup_channel: str | None = None, client: Valkey | None = None)[source]#

Bases: object

Configuration for the Valkey queue backend.

backend_name: ClassVar[str] = 'valkey'#

url: str#: Valkey connection URL used when no client is injected.

key_prefix: str | None#: Prefix applied to every queue key; None derives it from QueueConfig.namespace.

worker_wakeups: bool#: Whether workers subscribe for Valkey wakeup hints between polling passes.

__init__(url: str = 'redis://localhost:6379/0', key_prefix: str | None = None, worker_wakeups: bool = True, wakeup_channel: str | None = None, client: Valkey | None = None) → None#

wakeup_channel: str | None#: Worker-wakeup pub/sub channel; None derives it from QueueConfig.namespace.

client: Valkey | None#: Injected async Valkey client; None creates one from url.