Storage Backends¶

API reference for the authoritative queue storage contract (BaseQueueBackend), backend registry helpers, and all built-in storage backend configurations and implementations.

Backend Registry & Base Contract¶

get_queue_backend ¶

get_queue_backend(backend='memory', config=None)

Get an instantiated queue backend.

Returns:

Type	Description
`BaseQueueBackend`	A configured queue backend instance.

Raises:

Type	Description
`TypeError`	If a typed backend config selects a backend class that does not accept `backend_config`.

get_queue_backend_class ¶

get_queue_backend_class(backend_path)

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:

Type	Description
`type[BaseQueueBackend]`	The resolved queue backend class.

Raises:

Type	Description
`ValueError`	If a short backend name is unknown.

list_queue_backends ¶

list_queue_backends()

Return registered queue backend names (built-ins + dynamically registered).

queue_backend ¶

queue_backend(name)

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

Returns:

Type	Description
`Callable[[type[BaseQueueBackend]], type[BaseQueueBackend]]`	A decorator that registers the backend class.

BaseQueueBackend ¶

Base class for queue persistence backends.

open ¶

open()

Open queue resources.

Returns:

Type	Description
`bool`	True when resources are ready.

close ¶

close()

Close queue resources.

enqueue ¶

enqueue(
    task_name,
    *,
    args=(),
    kwargs=None,
    queue="default",
    priority=0,
    max_retries=0,
    scheduled_at=None,
    expires_at=None,
    key=None,
    execution_backend="local",
    execution_profile=None,
    metadata=None,
    id=None,
)

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.

enqueue_many ¶

enqueue_many(requests)

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:

Type	Description
`list[QueuedTaskRecord]`	Queue task records in the same order as `requests`.

claim_next ¶

claim_next(*, queues=(), execution_backend=None)

Claim the next due task across the requested queues.

An empty queues tuple claims across all queues.

Returns:

Type	Description
`QueuedTaskRecord \| None`	The claimed task record, if one was available.

claim_many ¶

claim_many(*, limit, queues=(), execution_backend=None, queue_limits=None)

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:

Type	Description
`list[QueuedTaskRecord]`	Claimed task records.

claim_task ¶

claim_task(task_id, *, expected_retry_count=None, expected_execution_ref=None)

Atomically claim a pending task.

complete_task ¶

complete_task(task_id, *, result=None, expected_retry_count=None)

Mark a task as completed.

Parameters:

Name	Type	Description	Default
`task_id`	`UUID`	Queue record identifier.	required
`result`	`Any`	Task result payload.	`None`
`expected_retry_count`	`int \| None`	When provided, update only if the record is still running with this retry count.	`None`

fail_task ¶

fail_task(
    task_id,
    error,
    *,
    retry=True,
    expected_retry_count=None,
    retry_at=None,
    queued_at=None,
)

Mark a task as failed or retry it.

Parameters:

Name	Type	Description	Default
`task_id`	`UUID`	Queue record identifier.	required
`error`	`str`	Error message to persist.	required
`retry`	`bool`	Whether retry policy may requeue the task.	`True`
`expected_retry_count`	`int \| None`	When provided, update only if the record is still running with this retry count.	`None`
`retry_at`	`datetime \| None`	Scheduled eligibility timestamp for a delayed retry.	`None`
`queued_at`	`datetime \| None`	Queue ordering timestamp for the new retry attempt.	`None`

cancel_task ¶

cancel_task(task_id, *, include_running=False, expected_retry_count=None)

Cancel a task.

Parameters:

Name	Type	Description	Default
`task_id`	`UUID`	ID of the task to cancel.	required
`include_running`	`bool`	Whether to cancel a task that is currently running.	`False`
`expected_retry_count`	`int \| None`	Only cancel if the task is still on this retry generation.	`None`

Returns:

Type	Description
`bool`	True if the task was successfully cancelled, False otherwise.

cancel_tasks ¶

cancel_tasks(
    *, task_name=None, queue=None, kwargs=None, metadata=None, include_running=False
)

Cancel tasks matching a domain predicate.

Parameters:

Name	Type	Description	Default
`task_name`	`str \| None`	Optional task name exact match.	`None`
`queue`	`str \| None`	Optional queue exact match.	`None`
`kwargs`	`Mapping[str, Any] \| None`	Optional top-level kwargs exact-match subset.	`None`
`metadata`	`Mapping[str, Any] \| None`	Optional top-level metadata exact-match subset.	`None`
`include_running`	`bool`	When true, running records are included for cooperative cancellation.	`False`

Returns:

Type	Description
`int`	Number of records cancelled.

requeue_stale_running ¶

requeue_stale_running(*, stale_after, limit=None)

Recover running tasks with stale heartbeats.

Parameters:

Name	Type	Description	Default
`stale_after`	`timedelta`	Heartbeat age past which a running task is stale.	required
`limit`	`int \| None`	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.	`None`

Returns:

Type	Description
`StaleTaskRecoveryResult`	Summary of requeued, failed, skipped, and handler-needed records.

Raises:

Type	Description
`NotImplementedError`	Always; every backend must answer this.

touch_heartbeats ¶

touch_heartbeats(touches)

Update heartbeat timestamps for running tasks.

Returns:

Type	Description
`HeartbeatTouchResult`	The task IDs confirmed touched or missed by the backend.

Raises:

Type	Description
`NotImplementedError`	Always; every backend must answer this.

cleanup_terminal ¶

cleanup_terminal(before, *, limit=None)

Delete terminal records completed before a cutoff.

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

Parameters:

Name	Type	Description	Default
`before`	`datetime`	Delete terminal records completed strictly before this UTC cutoff.	required
`limit`	`int \| None`	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.	`None`

Returns:

Type	Description
`int`	The number of deleted records.

Raises:

Type	Description
`NotImplementedError`	Always; every backend must answer this.

get_task ¶

get_task(task_id)

Return a queued task by ID.

get_tasks ¶

get_tasks(task_ids)

Return existing records for the supplied identifiers.

get_statistics ¶

get_statistics(*, queue=None)

Return queue status counts.

Raises:

Type	Description
`NotImplementedError`	Always; every backend must answer this.

SQLSpec Backend¶

SQLSpecBackendConfig ¶

Configuration values for the SQLSpec queue backend.

Attributes:

Name	Type	Description
`backend_name`	`str`	Registered queue storage backend identifier (`"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.
`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.
`terminal_ttl`	`float \| None`	Optional retention window in seconds for terminal task rows.

configure_migrations ¶

configure_migrations(config)

Register the queue and durable-events migrations with the SQLSpec config.

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

SQLSpecWorkerWakeupConfig ¶

SQLSpec worker-wakeup channel and transport configuration.

Attributes:

Name	Type	Description
`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.

SQLSpecQueueBackend ¶

Bases: BaseQueueBackend

SQLSpec-backed queue backend.

init ¶

__init__(config=None, *, backend_config=None)

open ¶

open()

Open SQLSpec resources.

Returns:

Type	Description
`bool`	True when SQLSpec resources are ready.

close ¶

close()

Close SQLSpec resources.

create_schema ¶

create_schema()

Create the SQLSpec queue table and indexes.

SQLSpecQueueStore ¶

Base SQLSpec queue statement store.

init ¶

__init__(
    config,
    *,
    table_name=None,
    column_map=None,
    native_json_columns=None,
    manage_schema=True,
    terminal_ttl=None,
)

create_queue_store ¶

create_queue_store(
    config,
    *,
    table_name=None,
    column_map=None,
    native_json_columns=None,
    manage_schema=True,
    terminal_ttl=None,
)

Create a queue store for a SQLSpec adapter configuration.

Returns:

Type	Description
`SQLSpecQueueStore`	The queue store implementation for the SQLSpec adapter.

configure_queue_migration_extension ¶

configure_queue_migration_extension(
    sqlspec_config,
    *,
    queue_table_name=DEFAULT_TABLE_NAME,
    event_history_enabled=False,
    event_history_table_name=None,
    event_history_extra_columns=(),
    maintenance_table_name=None,
    task_reservation_table_name=None,
    column_map=None,
    manage_schema=True,
    terminal_ttl=None,
)

Register the packaged queue migrations with SQLSpec's extension runner.

manage_schema=False declares that the application owns the queue schema, so nothing stays registered: no extension settings, no migration directory, and therefore no packaged revision for SQLSpec to discover or apply. A registration written by an earlier call on the same config is removed, so the postcondition holds regardless of call order.

Advanced Alchemy Backend¶

SQLAlchemyBackendConfig ¶

Configuration values for the SQLAlchemy queue backend.

Attributes:

Name	Type	Description
`backend_name`	`str`	Registered queue storage backend identifier (`"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.
`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.

SQLAlchemyBackend ¶

Bases: BaseQueueBackend

SQLAlchemy queue backend using Advanced Alchemy services.

init ¶

__init__(config=None, *, backend_config=None)

open ¶

open()

Open Advanced Alchemy resources.

Returns:

Type	Description
`bool`	True when resources are ready.

close ¶

close()

Close backend-owned resources.

QueueTaskModelMixin ¶

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 ¶

task_name()

Registered task callable identifier.

args_json ¶

args_json()

Serialized positional task arguments column (task_args).

kwargs_json ¶

kwargs_json()

Serialized keyword task arguments column (task_kwargs).

queue ¶

queue()

Logical queue name.

execution_backend ¶

execution_backend()

Execution backend identifier.

execution_profile ¶

execution_profile()

Optional execution profile identifier.

execution_ref ¶

execution_ref()

External execution reference or provisional reservation token.

worker_id ¶

worker_id()

Identifier of the worker owning the active attempt.

status ¶

status()

Current task lifecycle status.

priority ¶

priority()

Execution priority where larger numbers run first.

max_retries ¶

max_retries()

Maximum retry attempts allowed after the initial attempt.

retry_count ¶

retry_count()

Number of retries consumed or attempt fencing counter.

scheduled_at ¶

scheduled_at()

Earliest UTC timestamp at which the task is eligible to run.

expires_at ¶

expires_at()

UTC deadline after which an unclaimed task transitions to expired.

queued_at ¶

queued_at()

UTC timestamp when the current attempt entered the queue.

started_at ¶

started_at()

UTC timestamp when the active attempt was claimed.

completed_at ¶

completed_at()

UTC timestamp when the task reached a terminal status.

heartbeat_at ¶

heartbeat_at()

UTC timestamp of the latest worker heartbeat on a running attempt.

dispatch_checked_at ¶

dispatch_checked_at()

UTC timestamp when external dispatch repair last inspected the task.

result_json ¶

result_json()

Serialized task return value column (result).

error ¶

error()

Last failure message or stale-heartbeat error description.

task_key ¶

task_key()

Active uniqueness deduplication key.

metadata_json ¶

metadata_json()

Task metadata JSON dictionary column (metadata).

QueueEventHistoryModelMixin ¶

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

event_id ¶

event_id()

Unique queue event identifier.

event_type ¶

event_type()

Event type identifier such as task.completed.

task_id ¶

task_id()

Associated task UUID string when the event is task-scoped.

task_name ¶

task_name()

Registered task callable identifier.

queue ¶

queue()

Logical queue name.

worker_id ¶

worker_id()

Worker identifier that emitted the event.

execution_backend ¶

execution_backend()

Execution backend identifier associated with the event.

execution_profile ¶

execution_profile()

Optional execution profile identifier associated with the event.

actor_type ¶

actor_type()

Category of actor that emitted the event.

actor_id ¶

actor_id()

Identifier of the actor that emitted the event.

level ¶

level()

Severity level such as info, warning, or error.

message ¶

message()

Human-readable event message.

stage ¶

stage()

Logical execution stage label.

scope ¶

scope()

Application-defined logical event scope.

scope_key ¶

scope_key()

Application-defined scoping key for filtered queries.

actor ¶

actor()

Application-defined actor label.

entity ¶

entity()

Application-defined entity identifier.

detail_json ¶

detail_json()

Structured event payload dictionary.

progress_current ¶

progress_current()

Current progress value reported by the task.

progress_total ¶

progress_total()

Total progress target reported by the task.

progress_percent ¶

progress_percent()

Computed or explicit completion percentage.

duration_ms ¶

duration_ms()

Elapsed duration in milliseconds for terminal or stage events.

sequence ¶

sequence()

Monotonic per-task event sequence number.

occurred_at ¶

occurred_at()

UTC timestamp when the event occurred.

QueueMaintenanceModelMixin ¶

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 ¶

name()

Unique maintenance lock identifier.

token ¶

token()

Fencing token identifying the current maintenance lease holder.

expires_at ¶

expires_at()

UTC timestamp when the maintenance lease expires.

QueueTaskReservationModelMixin ¶

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 ¶

identity_key()

Unique forever-identity reservation key.

task_id ¶

task_id()

Originating task UUID string that claimed the reservation.

task_name ¶

task_name()

Registered task callable identifier that claimed the reservation.

Redis & Valkey Backends¶

RedisBackendConfig ¶

Configuration for the Redis queue backend.

Attributes:

Name	Type	Description
`backend_name`	`str`	Registered queue backend identifier (`"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.
`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`.
`terminal_ttl`	`float \| None`	Native TTL in seconds applied to terminal task records; `None` retains records until maintenance cleanup.

RedisQueueBackend ¶

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 ¶

__init__(config=None, *, backend_config=None)

open ¶

open()

Open Redis-protocol client resources.

Returns:

Type	Description
`bool`	True when the client is ready.

close ¶

close()

Drain history and attempt every owned resource cleanup.

ValkeyBackendConfig ¶

Configuration for the Valkey queue backend.

Attributes:

Name	Type	Description
`backend_name`	`str`	Registered queue backend identifier (`"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.
`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`.
`terminal_ttl`	`float \| None`	Optional retention window in seconds for terminal task hashes.

ValkeyQueueBackend ¶

Bases: RedisQueueBackend

Valkey-backed queue backend.

init ¶

__init__(config=None, *, backend_config=None)

Ephemeral & In-Memory Backends¶

EphemeralQueueBackend ¶

Bases: BaseQueueBackend

Private-file SQLite queue backend owned by one Litestar server invocation.

This backend is ephemeral, not durable. The database lives in a private temporary directory created and removed by the server lifespan, so queued work does not survive the server that produced it.

Worker wakeups are discovered by polling. There is no cross-process signalling primitive: a producer commits a row and a waiter notices it through short bounded reads, then the worker performs the authoritative transactional claim.

init ¶

__init__(config=None)

Initialize without resolving the database path.

The path is resolved in open(), because the top-level application constructs its backend before the server lifespan creates the database.

open ¶

open()

Resolve and validate the server-owned database.

Returns:

Type	Description
`bool`	True when the database is ready.

Raises:

Type	Description
`QueueConfigurationError`	If no server lifespan created the database, or the location, schema, or nonce does not match this invocation.

close ¶

close()

Cancel any retained notification wait.

InMemoryQueueBackend ¶

Bases: BaseQueueBackend

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

init ¶

__init__(config=None)

close ¶

close()

Cancel any retained notification and worker-control waits.

Storage Backends¶

Backend Registry & Base Contract¶

get_queue_backend ¶

get_queue_backend_class ¶

list_queue_backends ¶

queue_backend ¶

BaseQueueBackend ¶

open ¶

close ¶

enqueue ¶

enqueue_many ¶

claim_next ¶

claim_many ¶

claim_task ¶

complete_task ¶

fail_task ¶

cancel_task ¶

cancel_tasks ¶

requeue_stale_running ¶

touch_heartbeats ¶

cleanup_terminal ¶

get_task ¶

get_tasks ¶

get_statistics ¶

SQLSpec Backend¶

SQLSpecBackendConfig ¶

configure_migrations ¶

SQLSpecWorkerWakeupConfig ¶

SQLSpecQueueBackend ¶

__init__ ¶

open ¶

close ¶

create_schema ¶

SQLSpecQueueStore ¶

__init__ ¶

create_queue_store ¶

configure_queue_migration_extension ¶

Advanced Alchemy Backend¶

SQLAlchemyBackendConfig ¶

SQLAlchemyBackend ¶

__init__ ¶

open ¶

close ¶

QueueTaskModelMixin ¶

task_name ¶

args_json ¶

kwargs_json ¶

queue ¶

execution_backend ¶

execution_profile ¶

execution_ref ¶

worker_id ¶

status ¶

priority ¶

max_retries ¶

retry_count ¶

scheduled_at ¶

expires_at ¶

queued_at ¶

started_at ¶

completed_at ¶

heartbeat_at ¶

dispatch_checked_at ¶

result_json ¶

error ¶

task_key ¶

metadata_json ¶

QueueEventHistoryModelMixin ¶

event_id ¶

event_type ¶

task_id ¶

task_name ¶

queue ¶

worker_id ¶

execution_backend ¶

execution_profile ¶

actor_type ¶

actor_id ¶

level ¶

message ¶

init ¶

init ¶

init ¶

init ¶

init ¶

init ¶

init ¶