Tasks, Models & Exceptions¶

API reference for task registration, recurring schedule configuration, dependency hooks, persisted task models, and the exception hierarchy.

Tasks & Schedules¶

task ¶

task(
    func_or_name=None,
    /,
    *,
    queue="default",
    priority=0,
    retries=0,
    retry_backoff=None,
    timeout=None,
    execution_backend=None,
    execution_profile=None,
    key=None,
    unique_by=None,
    unique_until="terminal",
    run_after=None,
    expires_in=None,
    description=None,
    log_level=None,
    log_success=None,
    requeue_on_stale=None,
    requeue_on_shutdown=None,
    on_stale_failure=None,
    sync_to_thread=None,
    actor=None,
    cron=None,
    interval=None,
    timezone="UTC",
    initial_delay=0,
    jitter=0,
)

Register a callable as a queue task.

Returns:

Type	Description
`Task[Any, Any] \| Callable[[AnyTaskCallable], Task[Any, Any]]`	A task wrapper when used bare, otherwise a decorator.

Raises:

Type	Description
`ValueError`	If both cron and interval are configured.

Task ¶

Bases: Generic[P, T]

Registered task wrapper with direct call and enqueue APIs.

Attributes:

Name	Type	Description
`name`	`str`	Registered task name.
`queue`	`str`	Default queue name.
`priority`	`int`	Default priority.
`retries`	`int`	Maximum retry count.
`retry_backoff`	`RetryBackoff \| None`	Retry delay policy, if configured.
`timeout`	`float \| None`	Execution timeout.
`execution_backend`	`str \| None`	Task-specific execution backend override.
`execution_profile`	`str \| None`	Task-specific execution profile override.
`key`	`str \| None`	Default deduplication key.
`unique_by`	`TaskUniqueBy \| None`	Identity source when no explicit or configured key is supplied.
`unique_until`	`TaskUniqueUntil`	Identity lifetime: `terminal` (default) or `forever`.
`signature`	`inspect.Signature`	Cached call signature of the wrapped callable, computed once per task.
`run_after`	`timedelta \| None`	Relative delay for enqueue operations.
`expires_in`	`timedelta \| None`	Relative not-started deadline for enqueue operations.
`description`	`str \| None`	Task description metadata.
`log_level`	`str \| None`	Task log level metadata.
`log_success`	`bool \| None`	Whether successful completion logging should be quiet.
`requeue_on_stale`	`bool`	Whether stale running records should be requeued when retries remain.
`requeue_on_shutdown`	`bool \| None`	Task-specific shutdown requeue override.
`on_stale_failure`	`StaleFailureHandler \| None`	Callback invoked after this task reaches terminal stale failure.
`function`	`TaskCallable[P, T]`	Wrapped callable.
`sync_to_thread`	`bool \| None`	Whether a synchronous callable is offloaded to a worker thread.
`actor`	`QueueEventActor \| Callable[[], QueueEventActor] \| None`	Task-specific declared actor or actor resolver.

execute_record ¶

execute_record(record, *, task_context=None, extra_kwargs=None, sync_executor=None)

Execute this task for a queued record in worker context.

Returns:

Type	Description
`T`	The wrapped callable result.

metadata ¶

metadata(values=None)

Return enqueue metadata for this task.

using ¶

using(
    *,
    queue=None,
    priority=None,
    retries=None,
    retry_backoff=None,
    timeout=None,
    execution_backend=None,
    execution_profile=None,
    key=None,
    unique_by=None,
    unique_until=None,
    run_after=None,
    expires_in=None,
    description=None,
    log_level=None,
    log_success=None,
    requeue_on_stale=None,
    requeue_on_shutdown=None,
    on_stale_failure=None,
    sync_to_thread=None,
)

Return a configured copy with enqueue overrides.

enqueue ¶

enqueue(*args, **kwargs)

Enqueue this task using the configured default service or fall back to an immediate service.

Returns:

Type	Description
`TaskResult`	A result handle for the queued record.

ScheduleConfig ¶

Configuration for a recurring task schedule.

Attributes:

Name	Type	Description
`task_name`	`str`	Registered task name invoked by the schedule.
`cron`	`str \| None`	Five-field cron expression; `None` uses `interval` scheduling.
`interval`	`timedelta \| int \| float \| None`	Fixed interval as a duration or seconds; `None` uses `cron` scheduling.
`timezone`	`str`	IANA timezone used to evaluate the cron expression.
`initial_delay`	`timedelta \| int \| float`	Delay before the first interval run as a duration or seconds.
`jitter`	`timedelta \| int \| float`	Maximum random scheduling offset as a duration or seconds.

get_next_run ¶

get_next_run(after=None, *, use_initial_delay=False)

Calculate the next scheduled run time.

Returns:

Type	Description
`datetime`	The next run time in UTC.

Raises:

Type	Description
`ValueError`	If no interval or cron expression is configured.

as_metadata ¶

as_metadata()

Return a JSON-compatible metadata representation.

copy_for_task ¶

copy_for_task(task_name)

Return this normalized schedule for another task name.

RetryBackoff ¶

Delay policy applied when a task consumes a retry.

Attributes:

Name	Type	Description
`initial_delay`	`float`	Base retry delay in seconds for the first retry (`retry_count == 0`).
`multiplier`	`float`	Exponential growth multiplier (`>= 1.0`) applied per retry attempt.
`max_delay`	`float \| None`	Optional upper bound in seconds for any computed retry delay.

delay_for ¶

delay_for(retry_count)

Return the delay for a zero-based retry count.

as_metadata ¶

as_metadata()

Return the JSON-compatible persisted representation.

TaskResult ¶

Handle to a queued task result.

Attributes:

Name	Type	Description
`id`	`UUID`	Queue record ID.
`task_name`	`str`	Registered task name.
`status`	`TaskStatus \| None`	Cached task status.
`result`	`Any`	Cached task result.
`error`	`str \| None`	Cached task error.
`record`	`QueuedTaskRecord \| None`	Cached queue record.

refresh ¶

refresh()

Refresh this handle from its queue service.

Returns:

Type	Description
`Self`	The refreshed result handle.

Raises:

Type	Description
`RuntimeError`	If the result has no associated service.

wait ¶

wait(*, timeout=None, poll_interval=0.1)

Wait until the task reaches a terminal status.

Returns:

Type	Description
`Self`	The completed result handle.

Raises:

Type	Description
`TimeoutError`	If the timeout elapses before a terminal status.
`RuntimeError`	If the task no longer exists in the queue backend.

TaskUniqueBy ¶

TaskUniqueBy = Literal['task', 'arguments']

Identity source for task uniqueness: the registered task name or the normalized call arguments.

TaskUniqueUntil ¶

TaskUniqueUntil = Literal['terminal', 'forever']

Identity lifetime for task uniqueness: released at terminal (the default) or retained by a reservation.

TaskDependencyProvider ¶

Bases: Protocol

Per-attempt async context manager supplying task keyword arguments.

The queue enters this scope inside the attempt timeout, merges the yielded mapping into the task's keyword arguments, and awaits __aexit__ exactly once for every attempt outcome: success, retryable failure, terminal failure, timeout, cancellation, claim loss, and shutdown interruption.

__aexit__ receives asyncio.CancelledError for timeout, cancellation, claim loss, and shutdown alike, because all four reach the attempt as a coroutine cancellation. Read record and context to distinguish outcomes; never the exception type. A truthy __aexit__ return value is ignored: this is a resource scope, not an exception filter.

An implementation that also owns process-level state may expose open() and close() (synchronous or asynchronous); QueueService includes them in its own lifecycle and partial-open rollback.

TaskDependencyResolver ¶

TaskDependencyResolver = Callable[
    ["Task[..., object]", "QueuedTaskRecord", "TaskExecutionContext"],
    Awaitable[Mapping[str, object]],
]

User-supplied callable that resolves extra kwargs for a task before execution.

get_task_registry ¶

get_task_registry()

Return the global task registry.

get_scheduled_tasks ¶

get_scheduled_tasks()

Return the global scheduled task registry.

discover_tasks ¶

discover_tasks(package, subpackage='jobs', *, force_reload=False)

Walk package and import every <package>.<...>.<subpackage>.<...> module.

Adopters with app.domain.<x>.jobs/ layouts can call this once at startup so @task-decorated callables register without having to enumerate QueueConfig.task_modules by hand.

Parameters:

Name	Type	Description	Default
`package`	`str`	Dotted package name to walk (e.g. `"app.domain"`).	required
`subpackage`	`str`	Path segment that marks task modules. Any module whose dotted path (excluding the root) contains this segment is imported. Defaults to `"jobs"`.	`'jobs'`
`force_reload`	`bool`	Re-import modules already in `sys.modules`.	`False`

Returns:

Type	Description
`tuple[str, ...]`	Sorted, deduplicated tuple of task names registered after the walk.

Raises:

Type	Description
`ModuleNotFoundError`	If `package` cannot be imported, or if it resolves to a plain module rather than a package.

load_task_modules ¶

load_task_modules(modules, *, force_reload=False)

Import configured task modules so decorators register tasks.

Returns:

Type	Description
`int`	Number of imported modules.

Models & Statuses¶

QueuedTaskRecord ¶

Backend-neutral representation of a queued task.

Attributes:

Name	Type	Description
`task_name`	`str`	Registered task name to execute.
`id`	`UUID`	Unique identifier for this task record.
`args`	`tuple[Any, ...]`	Positional arguments persisted for the task invocation.
`kwargs`	`dict[str, Any]`	Keyword arguments persisted for the task invocation.
`queue`	`str`	Logical queue name on which the task is scheduled.
`execution_backend`	`str`	Name of the execution transport responsible for running the task.
`execution_profile`	`str \| None`	Optional execution profile name interpreted by the execution backend.
`execution_ref`	`str \| None`	External transport execution reference or reservation sentinel.
`worker_id`	`str \| None`	Identifier of the worker currently claiming or last executing the task.
`status`	`TaskStatus`	Current lifecycle status of the task record.
`priority`	`int`	Scheduling priority (higher integers are claimed before lower integers).
`max_retries`	`int`	Maximum number of retry attempts allowed on failure.
`retry_count`	`int`	Number of retries consumed so far; also serves as the lease fence counter.
`scheduled_at`	`datetime \| None`	UTC timestamp when a scheduled task becomes eligible for claiming.
`expires_at`	`datetime \| None`	UTC deadline after which an unstarted task transitions to `expired`.
`created_at`	`datetime`	UTC timestamp when the record object was first created.
`queued_at`	`datetime`	UTC timestamp when the record entered the queue or was last requeued.
`started_at`	`datetime \| None`	UTC timestamp when the current or most recent execution attempt started.
`completed_at`	`datetime \| None`	UTC timestamp when the task reached a terminal status.
`heartbeat_at`	`datetime \| None`	UTC timestamp of the most recent worker heartbeat while running.
`result`	`Any`	Deserialized return value from a completed task execution.
`error`	`str \| None`	Formatted exception message from the most recent failed attempt.
`key`	`str \| None`	Optional uniqueness or schedule identity key.
`metadata`	`dict[str, Any]`	Backend-neutral metadata dictionary (retry backoff, trace context, schedule info).
`dispatch_checked_at`	`datetime \| None`	UTC timestamp when external dispatch reconciliation last inspected this record.
`is_terminal`	`bool`	Whether the record is in a terminal state.
`is_due`	`bool`	Whether the record is eligible to be claimed now.
`is_expired`	`bool`	Whether the not-started deadline has passed.

TaskRequest ¶

A task submission for bulk enqueue via enqueue_many.

Carries the task name and enqueue arguments needed to submit one item in a batch. The backend returns the resulting queue records in request order.

Attributes:

Name	Type	Description
`task_name`	`str`	Registered task name to enqueue.
`args`	`tuple[Any, ...]`	Positional arguments serialized with the task record.
`kwargs`	`dict[str, Any] \| None`	Keyword arguments serialized with the task record.
`queue`	`str`	Target logical queue name.
`priority`	`int`	Relative task priority (higher values are claimed first).
`max_retries`	`int`	Maximum retry attempts permitted after an execution failure.
`scheduled_at`	`datetime \| None`	Optional UTC earliest execution timestamp.
`expires_at`	`datetime \| None`	Optional UTC deadline after which an unstarted task expires.
`key`	`str \| None`	Optional deduplication or uniqueness identity key.
`execution_backend`	`str`	Target execution transport name.
`execution_profile`	`str \| None`	Optional named execution profile for the target execution backend.
`metadata`	`dict[str, Any] \| None`	Optional JSON-serializable metadata dictionary stored with the record.

TaskReservation ¶

A durable unique_until="forever" identity reservation.

Records only the identity key, the originating task id/name, and the reservation time. It never carries args, kwargs, result, or error material. Routine terminal and event maintenance never delete it; an explicit administrative reset is the only removal path.

Attributes:

Name	Type	Description
`key`	`str`	Unique task identity key held permanently by this reservation.
`task_id`	`UUID`	Identifier of the task record that established the reservation.
`task_name`	`str`	Registered task name associated with the reservation.
`created_at`	`datetime`	UTC timestamp when the permanent identity reservation was recorded.

TaskStatus ¶

TaskStatus = Literal[
    "pending", "scheduled", "running", "completed", "failed", "cancelled", "expired"
]

Queue task lifecycle states.

QueueStatistics ¶

Operational status counts for a queue backend.

Attributes:

Name	Type	Description
`pending`	`int`	Count of due or immediate tasks waiting to be claimed.
`scheduled`	`int`	Count of future-scheduled tasks waiting for `scheduled_at`.
`running`	`int`	Count of tasks currently claimed by a worker or external execution transport.
`completed`	`int`	Count of tasks that finished successfully.
`failed`	`int`	Count of tasks that exhausted retries or failed permanently.
`cancelled`	`int`	Count of tasks cancelled before or during execution.
`expired`	`int`	Count of tasks whose `expires_at` deadline passed before execution started.
`total`	`int`	Total number of known queue records.

QueueBackendCapabilities ¶

Behavior advertised by a queue backend.

Attributes:

Name	Type	Description
`supports_worker_wakeups`	`bool`	Whether the backend emits push notifications when due work arrives.
`wakeup_backend`	`str \| None`	Name of the active wakeup notification transport, or `None` when polling.
`wakeups_durable`	`bool`	Whether wakeup notifications survive transient listener disconnections.
`supports_completion_events`	`bool`	Whether the backend emits push notifications when tasks reach terminal status.
`supports_maintenance`	`bool`	Whether the backend supports distributed maintenance leases and cleanup.

StaleTaskRecoveryResult ¶

Summary of stale running task recovery.

Attributes:

Name	Type	Description
`requeued`	`int`	Count of stale running tasks returned to `pending` or `scheduled`.
`failed`	`int`	Count of stale running tasks marked `failed` after exhausting retries.
`skipped`	`int`	Count of stale running tasks left untouched due to policy or active state.
`handler_needed`	`int`	Count of permanently failed stale tasks awaiting `on_stale_failure` callbacks.
`requeued_task_ids`	`list[UUID]`	Identifiers of tasks requeued during stale recovery.
`failed_task_ids`	`list[UUID]`	Identifiers of tasks transitioned to `failed` during stale recovery.
`handler_needed_task_ids`	`list[UUID]`	Identifiers of tasks requiring post-recovery failure handler invocation.
`repair`	`DispatchRepairResult \| None`	Structured external dispatch repair summary for requeued external tasks.

to_payload ¶

to_payload()

Return a JSON-compatible event payload.

ScheduleAssertionResult ¶

Summary of recurring schedule assertion and delivery repair.

Attributes:

Name	Type	Description
`created`	`int`	Count of missing recurring schedule records newly inserted.
`repaired`	`int`	Count of existing recurring schedule records whose external dispatch was repaired.
`repair`	`DispatchRepairResult \| None`	Structured external dispatch repair summary, if external repair ran.
`changed`	`int`	Total schedule occurrences created or repaired.

HeartbeatTouch ¶

A fenced heartbeat update request for one running task.

Attributes:

Name	Type	Description
`task_id`	`UUID`	Identifier of the running task whose lease is being extended.
`expected_retry_count`	`int \| None`	Retry-count fence verifying the active execution attempt.
`metadata_patch`	`dict[str, Any] \| None`	Optional key-value metadata patch merged during the heartbeat touch.

HeartbeatTouchResult ¶

Backend-neutral result for a bulk heartbeat update.

Attributes:

Name	Type	Description
`touched_task_ids`	`set[UUID]`	Identifiers of tasks whose heartbeat timestamps were updated.
`missed_task_ids`	`set[UUID]`	Identifiers of tasks that were no longer running or failed the fence check.

Exceptions & Terminal Helpers¶

non_retryable ¶

non_retryable(message)

Raise a non-retryable task failure.

Raises:

Type	Description
`NonRetryableError`	Always raised with the provided message.

job_cancelled ¶

job_cancelled(message='Task cancelled')

Raise a cooperative task cancellation.

Raises:

Type	Description
`JobCancelledError`	Always raised with the provided message.

QueueError ¶

Bases: Exception

Base exception for litestar-queues errors.

QueueWarning ¶

Bases: UserWarning

Base class for litestar-queues warnings.

QueueConfigurationError ¶

Bases: QueueError

Raised when queue backend configuration is invalid.

QueueDispatchError ¶

Bases: QueueError

Raised when a persisted record could not be handed to its transport.

committed is the part callers act on. A committed record is durable and a repair sweep can retry its dispatch; an uncommitted one never reached storage, so the caller owns retrying the whole enqueue.

QueueDispatchRepairError ¶

Bases: QueueError

Raised when a bounded reconciliation contains delivery repair failures.

TaskIdentityError ¶

Bases: QueueError

Raised when task uniqueness identity cannot be derived.

Signals that unique_by="arguments" was requested for a call whose bound arguments cannot be represented by the package's canonical JSON identity contract (for example non-finite floats or non-JSON objects). Uniqueness identity never falls back to pickle or repr(); the caller must supply an explicit key or pass identity-friendly arguments instead.

TaskIdentityTooLargeError ¶

Bases: QueueError

Raised when a canonical argument-identity payload exceeds the configured limit.

QueueEventBufferFull ¶

Bases: QueueError

Raised when queue event buffering cannot accept another event.

JobCancelledError ¶

Bases: QueueError

Raised by a task to cooperatively mark itself cancelled.

NonRetryableError ¶

Bases: QueueError

Raised by a task to mark the current failure as permanent.

MissingDependencyError ¶

Bases: QueueError, ImportError

Raised when a required optional dependency is not installed.