Plugin, Config & Service¶

API reference for the Litestar plugin, root configuration dataclasses, queue service, runtime namespace, and deferred post-response task adapter.

Plugin¶

QueuePlugin ¶

Bases: InitPlugin, CLIPlugin

Litestar plugin for queue service dependency registration and lifecycle.

Inheriting the concrete litestar.plugins.CLIPlugin is what makes Litestar register server_lifespan(); satisfying CLIPluginProtocol structurally is not enough.

Attributes:

Name	Type	Description
`config`	`QueueConfig`	Plugin configuration.

get_service ¶

get_service(state=None)

Return a QueueService for this plugin.

create_worker_service ¶

create_worker_service()

Create a fresh service that owns its process-local event resources.

standalone_app_lifespan ¶

standalone_app_lifespan(app=None, *, service=None)

Enter the application lifespan around a standalone worker, consumer, or maintenance service.

Parameters:

Name	Type	Description	Default
`app`	`Litestar \| None`	Optional Litestar application instance.	`None`
`service`	`QueueService \| None`	Optional pre-created `QueueService`; defaults to `create_worker_service()`.	`None`

Yields:

Type	Description
`AsyncIterator[QueueService]`	The active `QueueService` with `get_default_service()` bound.

open_worker_service ¶

open_worker_service(app=None)

Create a standalone worker service and enter the application lifespan when configured.

Parameters:

Name	Type	Description	Default
`app`	`Litestar \| None`	Optional Litestar application instance.	`None`

Yields:

Type	Description
`AsyncIterator[QueueService]`	The worker-owned `QueueService`.

on_app_init ¶

on_app_init(app_config)

Register queue dependencies, signature namespace, state, and the lifespan manager.

Returns:

Type	Description
`AppConfig`	The updated application configuration.

on_cli_init ¶

on_cli_init(cli)

Attach the queues subcommand group to the Litestar CLI.

Parameters:

Name	Type	Description	Default
`cli`	`Group`	The root `click.Group` of the Litestar CLI.	required

server_lifespan ¶

server_lifespan(app)

Own exactly one queue worker for the lifetime of a litestar run invocation.

Litestar enters this once, around the whole server command, for both its direct Uvicorn call and its multi-worker/reload subprocess path. Any alternative run-command plugin enters the same hook, so there is deliberately no server-specific detection or flag parsing here.

Yields:

Name	Type	Description
`None`	`Generator[None]`	with the invocation marker, storage, and worker child active.

Configuration¶

QueueConfig ¶

Configuration for QueuePlugin.

Attributes:

Name	Type	Description
`namespace`	`str`	Root used to derive package-owned runtime identifiers.
`queue_backend`	`QueueBackendConfig`	Queue-record persistence backend selector or typed backend configuration.
`execution_backend`	`ExecutionBackendConfig`	Default placement backend used to execute claimed tasks.
`execution_backends`	`Sequence[ExecutionBackendConfig] \| Mapping[str, ExecutionBackendConfig]`	Additional execution backend configurations available for task routing.
`task_dependency_resolver`	`TaskDependencyResolver \| None`	Per-attempt dependency resolver; `None` injects no additional task keyword arguments.
`task_dependency_provider`	`TaskDependencyProvider \| None`	Per-attempt dependency scope; `None` opens no scope around the task body.
`error_sanitizer`	`TaskErrorSanitizer \| None`	Persisted task-error formatter; `None` stores the default exception representation.
`worker`	`WorkerConfig`	Shared configuration for in-app and standalone workers.
`service_dependency_key`	`str \| None`	Litestar dependency key for the injected queue service; `None` derives it from the namespace.
`events_dependency_key`	`str \| None`	Litestar dependency key for the event producer; `None` derives it from the namespace.
`events`	`QueueEventsConfig \| None`	Task-event capabilities; `None` disables delivery, streams, and history.
`observability`	`ObservabilityConfig \| None`	Package telemetry configuration; `None` disables the observability runtime.
`task_modules`	`tuple[str, ...]`	Module names imported during startup to register decorated tasks.
`initialize_schedules`	`bool`	Whether application startup synchronizes registered recurring schedules.
`log_success`	`bool`	Whether successful task completion emits an informational log by default.
`sync_thread_pool_size`	`int`	Maximum threads running synchronous tasks concurrently.
`sync_thread_name_prefix`	`str \| None`	Thread-name prefix for synchronous tasks; `None` derives it from the namespace.
`scheduler_canary_task`	`str \| None`	Scheduler-health task name; `None` derives the package-owned default.
`maintenance`	`QueueMaintenanceConfig \| None`	Automatic maintenance policy; `None` disables the maintenance loop.
`max_argument_identity_bytes`	`int \| None`	Maximum canonical argument-identity size in bytes; `None` disables the bound.
`stale_requeue_priority`	`StaleRequeuePriority`	Priority applied to stale-recovered work; see `StaleRequeuePriority`.
`names`	`QueueNamespace`	Validated format-specific runtime-name renderer.
`signature_namespace`	`dict[str, Any]`	Names Litestar must resolve that an application cannot supply itself.
`dependencies`	`dict[str, Any]`	Dependency providers for Litestar's DI system.
`service_state_key`	`str`	Litestar state key holding the queue service.
`worker_state_key`	`str`	Litestar state key holding the in-process worker.
`event_publisher_state_key`	`str`	Litestar state key holding the event publisher.
`event_channels_state_key`	`str`	Litestar state key holding the resolved channels backend.
`observability_runtime_state_key`	`str`	Litestar state key holding the observability runtime.
`maintenance_name`	`str`	Distributed maintenance coordination name.

resolved_execution_backends ¶

resolved_execution_backends()

Return all configured execution backends keyed by registered backend name.

Raises:

Type	Description
`QueueConfigurationError`	If a mapping key does not match its backend name or conflicting configurations are provided for the same backend.

get_service ¶

get_service(state=None)

Return a QueueService for this configuration.

get_queue_backend ¶

get_queue_backend()

Return a configured queue backend instance.

get_execution_backend ¶

get_execution_backend(name=None)

Return a configured execution backend instance.

get_event_publisher ¶

get_event_publisher(*, channels_backend=None, manage_channels_lifecycle=False)

Return a configured queue event publisher.

Parameters:

Name	Type	Description	Default
`channels_backend`	`ChannelsLike \| None`	Fallback live sink target used only when `QueueEventsConfig.channels` is unset. `QueuePlugin` passes the app's registered `ChannelsPlugin` here so event delivery needs no manual channel wiring.	`None`
`manage_channels_lifecycle`	`bool`	Whether the publisher owns the resolved Channels target lifecycle.	`False`

provide_service_dependency ¶

provide_service_dependency(state)

Yield the application-scoped QueueService for Litestar dependency injection.

provide_event_producer_dependency ¶

provide_event_producer_dependency(state)

Yield the application-scoped QueueEventProducer for Litestar dependency injection.

WorkerConfig ¶

Configuration shared by in-app and standalone workers.

Attributes:

Name	Type	Description
`placement`	`WorkerPlacement`	Which process owns this worker; see `WorkerPlacement`.
`id`	`str \| None`	Explicit worker identity; `None` uses a process-derived identifier.
`batch_size`	`int`	Maximum task records claimed in one worker iteration.
`poll_interval`	`float`	Base worker polling interval in seconds.
`poll_backoff_max`	`float \| None`	Maximum adaptive polling interval in seconds; `None` disables backoff.
`poll_backoff_multiplier`	`float`	Multiplier applied after an empty polling iteration.
`poll_jitter`	`float`	Symmetric polling jitter ratio from zero through one.
`max_concurrency`	`int`	Maximum number of tasks executed concurrently.
`queue_concurrency`	`Mapping[str, int]`	Per-worker concurrency caps for named queues.
`heartbeat_interval`	`float`	Interval between bulk heartbeat writes in seconds.
`heartbeat_jitter_fraction`	`float`	Positive heartbeat delay jitter ratio from zero through one.
`heartbeat_miss_threshold`	`int`	Consecutive heartbeat misses tolerated before claim loss.
`cancel_on_claim_loss`	`bool`	Whether heartbeat claim loss cancels the locally running coroutine.
`cancellation_poll_interval`	`float`	Interval between durable running-cancellation reconciliation passes.
`reconcile_interval`	`float`	Interval between external-execution reconciliation passes in seconds.
`stale_after`	`float \| None`	Running-task age threshold in seconds; `None` disables stale recovery.
`stale_check_interval`	`float`	Interval between stale-task recovery passes in seconds.
`expiry_check_interval`	`float \| None`	Interval between pending-job expiration passes; `None` disables sweeps.
`terminal_retention`	`float \| None`	Age threshold in seconds for in-worker terminal-record cleanup; `None` inherits backend `terminal_ttl` when set.
`terminal_check_interval`	`float`	Interval between in-worker terminal-record cleanup passes in seconds.
`graceful_shutdown_timeout`	`float`	Maximum graceful drain time in seconds.
`final_cancel_timeout`	`float`	Maximum post-cancellation drain time in seconds.
`hard_exit_timeout`	`float \| None`	Wall-clock budget from forced shutdown to process exit; `None` disables the watchdog.
`requeue_on_shutdown`	`bool`	Whether cancelled executions are requeued after shutdown drain timeout.
`max_interruptions`	`int`	Shutdown requeues an attempt may absorb before interruptions consume the retry budget.
`startup_timeout`	`float`	Maximum time to wait for worker startup readiness in seconds.
`queues`	`tuple[str, ...]`	Queue names claimed by this worker; empty claims every queue.
`run_app_lifespan`	`bool`	Whether standalone worker, consumer, and maintenance processes enter the Litestar application lifespan.

QueueMaintenanceConfig ¶

Bounded maintenance thresholds and limits.

Durations and retention values are seconds. Every limit and duration must be positive and coordination_timeout must exceed time_budget so ownership outlives the whole run. stale_after, terminal_retention, and event_retention_rules default to None (or empty) which disables their phase; there are no destructive defaults.

Attributes:

Name	Type	Description
`time_budget`	`float`	Budget checked before each phase; in-flight operations can run longer.
`coordination_timeout`	`float`	Distributed ownership duration in seconds; must exceed `time_budget`.
`expire_limit`	`int`	Maximum overdue pending or scheduled tasks expired in one run.
`external_limit`	`int`	Maximum external executions reconciled in one run.
`stale_after`	`float \| None`	Running-task age threshold in seconds; `None` disables stale recovery.
`stale_limit`	`int`	Maximum stale running tasks recovered in one run.
`schedules_limit`	`int`	Maximum recurring schedules asserted in one run.
`terminal_retention`	`float \| None`	Terminal-task retention age in seconds; `None` disables deletion.
`terminal_limit`	`int`	Maximum expired terminal tasks deleted in one run.
`event_retention_rules`	`tuple[QueueEventRetentionRule, ...]`	Ordered event-history retention rules; empty disables the events phase.
`event_limit`	`int`	Maximum expired task-event records deleted in one run.
`mount_route`	`bool`	Whether `QueuePlugin` mounts an HTTP endpoint that runs maintenance once.
`route_path`	`str`	Path the HTTP maintenance endpoint is mounted on.
`guards`	`tuple[Guard, ...]`	Application guards applied to the HTTP maintenance endpoint.
`trust_platform_auth`	`bool`	Whether platform IAM alone is accepted as protection for the HTTP maintenance endpoint.
`verify_oidc`	`bool`	Whether the HTTP maintenance endpoint verifies Google OIDC Bearer tokens in-process.
`oidc_service_account_email`	`str \| None`	Expected Google OIDC service account email when `verify_oidc=True`.
`oidc_audience`	`str \| None`	Expected Google OIDC audience when `verify_oidc=True`.

resolve ¶

resolve(namespace=None)

Resolve namespace-owned defaults without mutating this reusable config.

Service¶

QueueService ¶

High-level facade for queue and execution backends.

Attributes:

Name	Type	Description
`config`	`QueueConfig`	Queue configuration.
`observability_runtime`	`QueueObservabilityRuntimeProtocol`	Return the configured observability runtime.

get_queue_backend ¶

get_queue_backend()

Return the configured queue backend.

get_execution_backend ¶

get_execution_backend(name=None)

Return the configured execution backend for name (or the default backend).

get_execution_backends ¶

get_execution_backends()

Return all configured execution backends keyed by backend name.

get_event_publisher ¶

get_event_publisher()

Return the configured event publisher.

get_event_producer ¶

get_event_producer()

Return a producer over this service's event publisher.

get_event_log ¶

get_event_log()

Return the backend-owned durable event history, if configured.

History is wired during open() only when events.history is present; otherwise this returns None and history maintenance is a no-op.

open ¶

open()

Open queue and execution backends.

If the configured task dependency provider exposes an open() method, it is called first. If the provider fails to open, it will not be closed.

Returns:

Type	Description
`Self`	The opened service.

close ¶

close()

Close queue and execution backends.

enqueue ¶

enqueue(
    task,
    *args,
    scheduled_at=None,
    run_after=None,
    expires_in=None,
    expires_at=None,
    key=None,
    queue=None,
    priority=None,
    retries=None,
    timeout=None,
    execution_backend=None,
    execution_profile=None,
    description=None,
    log_level=None,
    log_success=None,
    requeue_on_stale=None,
    metadata=None,
    **kwargs,
)

Enqueue a registered task.

Returns:

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

schedule_persisted ¶

schedule_persisted(record)

Hand an already-persisted record to a self-scheduling execution backend.

A no-op for every polled backend, so persistence paths can call it unconditionally. The record is reloaded afterwards because the backend may have written a delivery reference the caller would otherwise miss.

Returns:

Type	Description
`QueuedTaskRecord`	The live record.

Raises:

Type	Description
`QueueDispatchError`	If dispatch or the final reload fails after the record was committed. The error retains the original task ID.

resolve_task ¶

resolve_task(task)

Resolve a task name or wrapper to a registered task.

Returns:

Type	Description
`Task[Any, Any]`	The registered task wrapper.

Raises:

Type	Description
`KeyError`	If a task name is not registered.

get_task ¶

get_task(task_id)

Return a queued task record by ID.

get_statistics ¶

get_statistics(*, queue=None)

Return global or queue-scoped task status counts.

cancel_task ¶

cancel_task(task_id, *, include_running=False)

Cancel one queued task and publish its terminal lifecycle event.

Parameters:

Name	Type	Description	Default
`task_id`	`UUID`	Identifier of the task to cancel.	required
`include_running`	`bool`	Whether a running task may transition to cancelled.	`False`

Returns:

Type	Description
`bool`	`True` only for the caller that wins the durable state transition.

interrupt_task ¶

interrupt_task(record, *, worker_id, reason='shutdown', max_interruptions=None)

Requeue one owned running attempt after local execution unwinds.

An attempt that keeps being interrupted would otherwise cycle forever without ever consuming its retry budget, so at max_interruptions the interruption is routed through the ordinary retry policy instead.

Returns:

Type	Description
`QueuedTaskRecord \| None`	The requeued or failed record, or `None` when the fence was lost.

reset_task_identity ¶

reset_task_identity(key)

Delete a unique_until="forever" reservation by its exact effective key.

This is the only supported way to allow a forever identity to be enqueued again. It never infers or resets an identity from raw arguments; the caller must pass the exact effective key (for example the lq:u:v1:... value or the configured/explicit key).

Returns:

Type	Description
`bool`	`True` when a reservation was removed.

get_task_identity ¶

get_task_identity(key)

Return the forever reservation owning an identity key, if any.

execute_record ¶

execute_record(record, *, worker_id=None)

Execute a claimed queue record and persist the lifecycle result.

Parameters:

Name	Type	Description	Default
`record`	`QueuedTaskRecord`	The claimed queue record to execute.	required
`worker_id`	`str \| None`	Identity of the worker driving execution, if any. The value is forwarded to `TaskExecutionContext.worker_id` so published events carry stable worker provenance. Service-driven executions (no worker) leave this as `None`.	`None`

Returns:

Type	Description
`QueuedTaskRecord`	The updated queue record.

Raises:

Type	Description
`CancelledError`	If task execution is cancelled.

reconcile_external ¶

reconcile_external(limit=None)

Reconcile externally dispatched records against their executor.

A bounded call first asks the configured backend to repair deliveries its transport lost, then spends what is left of the budget on ordinary reconciliation, so one maintenance pass stays finite however the two divide the work. The worker's unbounded sweep skips repair: it has no ceiling to respect, and repair is a maintenance responsibility.

Parameters:

Name	Type	Description	Default
`limit`	`int \| None`	When provided, examine at most this many external records across both halves. `None` reconciles every outstanding external record and repairs nothing.	`None`

Returns:

Type	Description
`int`	Number of records repaired or brought to a terminal queue status.

Raises:

Type	Description
`QueueDispatchRepairError`	If any bounded repair failed. The error carries the structured result, including successful changes.

reconcile_external_result ¶

reconcile_external_result(*, limit)

Repair and reconcile external records within one shared allowance.

A zero allowance returns a deferred result without accessing storage or the execution backend. Exhausting a repair page conservatively signals more work may remain; it does not count a backlog. Provider calls are not interrupted by the maintenance phase's time budget.

Returns:

Type	Description
`ExternalReconciliationResult`	Repair counts and the number of records reconciled afterward.

Raises:

Type	Description
`QueueConfigurationError`	If the allowance is negative.

recover_stale_tasks ¶

recover_stale_tasks(*, stale_after, worker_id=None, limit=None, redispatch=True)

Recover stale running tasks and publish a worker summary event.

Parameters:

Name	Type	Description	Default
`stale_after`	`timedelta`	Heartbeat age past which a running task is stale.	required
`worker_id`	`str \| None`	Identity attached to published recovery events.	`None`
`limit`	`int \| None`	When provided, recover at most this many records in one bounded batch. `None` preserves the unbounded worker behavior.	`None`
`redispatch`	`bool`	Whether to immediately schedule requeued records on self-scheduling execution backends.	`True`

Returns:

Type	Description
`StaleTaskRecoveryResult`	Summary of recovered, failed, skipped, and handler-needed tasks.

expire_overdue_tasks ¶

expire_overdue_tasks(*, limit=None, worker_id=None)

Expire overdue pending or scheduled records and publish one event each.

Returns:

Type	Description
`list[QueuedTaskRecord]`	Records transitioned to `expired`.

claim_tasks ¶

claim_tasks(
    *, limit, queues=(), execution_backend=None, worker_id=None, queue_limits=None
)

Claim due tasks and publish events for claim-time expirations.

Returns:

Type	Description
`list[QueuedTaskRecord]`	Records successfully transitioned to `running`.

claim_task ¶

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

Claim one task and publish its claim-time expiration event.

Returns:

Type	Description
`tuple[QueuedTaskRecord \| None, QueuedTaskRecord \| None]`	The claimed record and the expired record, at most one of which is set.

initialize_schedules ¶

initialize_schedules()

Create queue records for registered recurring schedules.

New occurrences snapshot the registered retry budget. Reusing a nonterminal occurrence preserves its existing budget and backoff.

Returns:

Type	Description
`list[QueuedTaskRecord]`	The created or reused schedule records.

has_registered_schedules ¶

has_registered_schedules()

Return whether recurring schedules are enabled and registered.

reassert_schedules ¶

reassert_schedules(*, limit=None)

Create missing recurring schedule occurrences and repair undispatched active ones.

Returns:

Type	Description
`ScheduleAssertionResult`	Counts of created and repaired schedule occurrences plus dispatch repair details.

Raises:

Type	Description
`QueueConfigurationError`	If `limit` is negative.

publish_claim_lost ¶

publish_claim_lost(
    record, *, phase, task_context=None, worker_id=None, expected_retry_count=None
)

Publish an ownership-loss event and return the current record state.

Returns:

Type	Description
`QueuedTaskRecord`	Current queue task record state.

get_default_service ¶

get_default_service()

Return the global default QueueService instance.

set_default_service ¶

set_default_service(service)

Set the global default QueueService instance.

Runtime Namespace & Background Tasks¶

QueueNamespace ¶

Validated root for package-owned runtime identifiers.

Attributes:

Name	Type	Description
`root`	`str`	Root namespace identifier prefix for package-owned runtime resources.
`is_default`	`bool`	Return whether this root is the compatibility namespace.

metric ¶

metric(*parts)

Render an OpenTelemetry or Prometheus identifier.

logger ¶

logger(*parts)

Render a runtime logger name.

channel ¶

channel(*parts)

Render a pub/sub or event channel.

key ¶

key(*parts)

Render a storage key.

database_channel ¶

database_channel(*parts)

Render a database notification channel.

registration ¶

registration(*parts)

Render a Litestar state, dependency, or route registration.

environment ¶

environment(*parts)

Render an environment-variable name.

resource ¶

resource(*parts)

Render a process, thread, or filesystem resource name.

coordination ¶

coordination(*parts)

Render a distributed coordination name with legacy compatibility.

package_task ¶

package_task(*parts)

Render a package-owned built-in task name with legacy compatibility.

QueuedBackgroundTask ¶

Bases: BackgroundTask

Background task that enqueues a queue task after the response is sent.