Core API#

Configuration#

litestar_queues.config.STALE_REQUEUE_PRIORITY: Literal['preserve'] | int | Callable[[int], int] = 'preserve'#

recovered work keeps the priority it was enqueued with.

Type:: Default stale-requeue policy

litestar_queues.config.ExecutionBackendConfig = str | litestar_queues.config.ExecutionBackendConfigProtocol#: Type alias for execution backend selectors.

class litestar_queues.config.ExecutionBackendConfigProtocol(*args, **kwargs)[source]#

Bases: Protocol

Protocol for typed execution backend configuration objects.

backend_name: ClassVar[str]#

__init__(*args, **kwargs)#

class litestar_queues.config.MigrationConfiguringBackend(*args, **kwargs)[source]#

Bases: Protocol

Backend config that registers its own migrations during plugin init.

A backend owns whatever application wiring its storage needs. QueuePlugin only asks whether the configured backend provides this hook, so selecting one backend never imports another backend’s package – or requires its extra to be installed.

configure_migrations(config: QueueConfig) → None[source]#: Register backend-owned migrations with the application.

__init__(*args, **kwargs)#

litestar_queues.config.QueueBackendConfig = str | litestar_queues.config.QueueBackendConfigProtocol#: Type alias for queue backend selectors.

class litestar_queues.config.QueueBackendConfigProtocol(*args, **kwargs)[source]#

Bases: Protocol

Protocol for typed queue backend configuration objects.

backend_name: ClassVar[str]#

__init__(*args, **kwargs)#

class litestar_queues.config.QueueConfig(namespace: str = 'litestar_queues', queue_backend: QueueBackendConfig = 'ephemeral', execution_backend: ExecutionBackendConfig = 'local', task_dependency_resolver: TaskDependencyResolver | None = None, task_dependency_provider: TaskDependencyProvider | None = None, error_sanitizer: TaskErrorSanitizer | None = None, worker: WorkerConfig = <factory>, service_dependency_key: str | None = None, events_dependency_key: str | None = None, events: QueueEventsConfig | None = None, observability: ObservabilityConfig | None = None, task_modules: tuple[str, ...] = (), initialize_schedules: bool = True, log_success: bool = False, sync_thread_pool_size: int = <factory>, sync_thread_name_prefix: str | None = None, scheduler_canary_task: str | None = None, maintenance: QueueMaintenanceConfig | None = None, max_argument_identity_bytes: int | None = None, stale_requeue_priority: StaleRequeuePriority = 'preserve')[source]#

Bases: object

Configuration for QueuePlugin.

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.

task_dependency_resolver: TaskDependencyResolver | None#: Per-attempt dependency resolver; None injects no additional task keyword arguments.

__init__(namespace: str = 'litestar_queues', queue_backend: QueueBackendConfig = 'ephemeral', execution_backend: ExecutionBackendConfig = 'local', task_dependency_resolver: TaskDependencyResolver | None = None, task_dependency_provider: TaskDependencyProvider | None = None, error_sanitizer: TaskErrorSanitizer | None = None, worker: WorkerConfig = <factory>, service_dependency_key: str | None = None, events_dependency_key: str | None = None, events: QueueEventsConfig | None = None, observability: ObservabilityConfig | None = None, task_modules: tuple[str, ...] = (), initialize_schedules: bool = True, log_success: bool = False, sync_thread_pool_size: int = <factory>, sync_thread_name_prefix: str | None = None, scheduler_canary_task: str | None = None, maintenance: QueueMaintenanceConfig | None = None, max_argument_identity_bytes: int | None = None, stale_requeue_priority: StaleRequeuePriority = 'preserve') → None#

task_dependency_provider: TaskDependencyProvider | None#

Per-attempt dependency scope; None opens no scope around the task body.

Mutually exclusive with task_dependency_resolver. Use the resolver for stateless keyword arguments and the provider when the attempt must own a resource that has to be released on every outcome.

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.

Defaults to the cgroup-aware effective CPU count plus four, capped at 32. Threads are created on demand, so this is a ceiling rather than a startup cost.

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.

property signature_namespace: dict[str, Any]#

Names Litestar must resolve that an application cannot supply itself.

This carries only the types named in this package’s own dependency providers. provide_service_dependency and provide_event_producer_dependency annotate their return types as strings while importing those types under TYPE_CHECKING, so Litestar needs them here to resolve the injected queue_service and queue_events dependencies.

Nothing else belongs here. Config, backend, worker, and event types are named in application setup code, not in handler signatures, and a handler that does annotate one has already imported it. Registering the whole public API instead made QueuePlugin.on_app_init import every installed adapter on every application startup, which defeated the package’s lazy-import boundary and charged applications for extras they never selected.

property dependencies: dict[str, Any]#: Dependency providers for Litestar’s DI system.

property service_state_key: str#: Litestar state key holding the queue service.

property worker_state_key: str#: Litestar state key holding the in-process worker.

property event_publisher_state_key: str#: Litestar state key holding the event publisher.

property event_channels_state_key: str#: Litestar state key holding the resolved channels backend.

property observability_runtime_state_key: str#: Litestar state key holding the observability runtime.

property maintenance_name: str#: Distributed maintenance coordination name.

get_service(state: State | None = None) → QueueService[source]#: Return a QueueService for this configuration.

get_queue_backend() → BaseQueueBackend[source]#: Return a configured queue backend instance.

get_execution_backend() → BaseExecutionBackend[source]#: Return a configured execution backend instance.

get_event_publisher(*, channels_backend: ChannelsLike | None = None, manage_channels_lifecycle: bool = False) → QueueEventPublisher[source]#

Return a configured queue event publisher.

Parameters:

channels_backend – 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.
manage_channels_lifecycle – Whether the publisher owns the resolved Channels target lifecycle.

async provide_service_dependency(state: State) → AsyncIterator['QueueService'][source]#: Yield the application-scoped QueueService for Litestar dependency injection.

async provide_event_producer_dependency(state: State) → AsyncIterator[QueueEventProducer][source]#: Yield the application-scoped QueueEventProducer for Litestar dependency injection.

class litestar_queues.config.QueueNamespace(root: str = 'litestar_queues')[source]#

Bases: object

Validated root for package-owned runtime identifiers.

root: str#

metric(*parts: str) → str[source]#: Render an OpenTelemetry or Prometheus identifier.

logger(*parts: str) → str[source]#: Render a runtime logger name.

channel(*parts: str) → str[source]#: Render a pub/sub or event channel.

key(*parts: str) → str[source]#: Render a storage key.

database_channel(*parts: str) → str[source]#: Render a database notification channel.

registration(*parts: str) → str[source]#: Render a Litestar state, dependency, or route registration.

environment(*parts: str) → str[source]#: Render an environment-variable name.

resource(*parts: str) → str[source]#: Render a process, thread, or filesystem resource name.

coordination(*parts: str) → str[source]#: Render a distributed coordination name with legacy compatibility.

package_task(*parts: str) → str[source]#: Render a package-owned built-in task name with legacy compatibility.

property is_default: bool#: Return whether this root is the compatibility namespace.

__init__(root: str = 'litestar_queues') → None#

litestar_queues.config.StaleRequeuePriority#

Priority policy applied to work recovered by the stale sweep.

"preserve" keeps the original priority, an int is a ceiling clamp, and a callable maps the current priority to the recovered one.

alias of Literal[‘preserve’] | int | Callable[[int], int]

class litestar_queues.config.TaskDependencyProvider(*args, **kwargs)[source]#

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.

__init__(*args, **kwargs)#

litestar_queues.config.TaskDependencyResolver#: User-supplied callable that resolves extra kwargs for a task before execution.

litestar_queues.config.TaskErrorSanitizer#: User-supplied callable that converts task exceptions into persisted error messages.

class litestar_queues.config.WorkerConfig(placement: ~typing.Literal['server', 'asgi', 'external'] = 'server', id: str | None = None, batch_size: int = 10, poll_interval: float = 0.1, poll_backoff_max: float | None = 30.0, poll_backoff_multiplier: float = 2.0, poll_jitter: float = 0.15, max_concurrency: int = 1, queue_concurrency: ~collections.abc.Mapping[str, int] = <factory>, heartbeat_interval: float = 30, heartbeat_jitter_fraction: float = 0.1, heartbeat_miss_threshold: int = 2, cancel_on_claim_loss: bool = True, cancellation_poll_interval: float = 1.0, reconcile_interval: float = 30, stale_after: float | None = None, stale_check_interval: float = 60.0, expiry_check_interval: float | None = 60.0, graceful_shutdown_timeout: float = 30, final_cancel_timeout: float = 5, hard_exit_timeout: float | None = 10.0, requeue_on_shutdown: bool = False, max_interruptions: int = 3, startup_timeout: float = 30, queues: tuple[str, ...] = ())[source]#

Bases: object

Configuration shared by in-app and standalone workers.

placement: Literal['server', 'asgi', 'external']#: 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.

Writes are already fenced on the claim, but side effects are not: while a lost claim keeps running, a replacement worker executes the same record. Cancelling closes that duplicate-execution window. Set to False to let a lost attempt run to completion and have its terminal write rejected.

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.

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.

__init__(placement: ~typing.Literal['server', 'asgi', 'external'] = 'server', id: str | None = None, batch_size: int = 10, poll_interval: float = 0.1, poll_backoff_max: float | None = 30.0, poll_backoff_multiplier: float = 2.0, poll_jitter: float = 0.15, max_concurrency: int = 1, queue_concurrency: ~collections.abc.Mapping[str, int] = <factory>, heartbeat_interval: float = 30, heartbeat_jitter_fraction: float = 0.1, heartbeat_miss_threshold: int = 2, cancel_on_claim_loss: bool = True, cancellation_poll_interval: float = 1.0, reconcile_interval: float = 30, stale_after: float | None = None, stale_check_interval: float = 60.0, expiry_check_interval: float | None = 60.0, graceful_shutdown_timeout: float = 30, final_cancel_timeout: float = 5, hard_exit_timeout: float | None = 10.0, requeue_on_shutdown: bool = False, max_interruptions: int = 3, startup_timeout: float = 30, queues: tuple[str, ...] = ()) → None#

litestar_queues.config.WorkerPlacement#

Which process owns the queue worker.

server: The Litestar CLI server lifespan owns exactly one fresh worker process per litestar run invocation. This is the default.
asgi: Each ASGI worker owns one queue worker inside its own application lifespan. Deliberately multiplicative with the web-worker count.
external: Nothing is started automatically; a separate process manager runs litestar queues run, or the caller executes tasks inline.

alias of Literal[‘server’, ‘asgi’, ‘external’]

litestar_queues.config.execution_backend_name(backend: str | ExecutionBackendConfigProtocol) → str[source]#: Return the registered execution backend name for a selector.

litestar_queues.config.queue_backend_name(backend: str | QueueBackendConfigProtocol) → str[source]#: Return the registered queue backend name for a selector.

Service#

class litestar_queues.service.QueueService(config: QueueConfig, *, queue_backend: BaseQueueBackend | None = None, execution_backend: BaseExecutionBackend | None = None, event_publisher: QueueEventPublisher | None = None, observability_runtime: QueueObservabilityRuntimeProtocol | None = None)[source]#

Bases: object

High-level facade for queue and execution backends.

__init__(config: QueueConfig, *, queue_backend: BaseQueueBackend | None = None, execution_backend: BaseExecutionBackend | None = None, event_publisher: QueueEventPublisher | None = None, observability_runtime: QueueObservabilityRuntimeProtocol | None = None) → None[source]#: Initialize the queue service.

property config: QueueConfig#: Queue configuration.

get_queue_backend() → BaseQueueBackend[source]#: Return the configured queue backend.

get_execution_backend() → BaseExecutionBackend[source]#: Return the configured execution backend.

get_event_publisher() → QueueEventPublisher[source]#: Return the configured event publisher.

get_event_producer() → QueueEventProducer[source]#: Return a producer over this service’s event publisher.

get_event_log() → QueueEventLog | None[source]#

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.

property observability_runtime: QueueObservabilityRuntimeProtocol#: Return the configured observability runtime.

async open() → Self[source]#

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:: The opened service.

async close() → None[source]#: Close queue and execution backends.

Enqueue a registered task.

Returns:: A result handle for the queued record.

resolve_task(task: str | Task[Any, Any]) → Task[Any, Any][source]#

Resolve a task name or wrapper to a registered task.

Returns:: The registered task wrapper.
Raises:: KeyError – If a task name is not registered.

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

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

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

Cancel one queued task and publish its terminal lifecycle event.

Parameters:

task_id – Identifier of the task to cancel.
include_running – Whether a running task may transition to cancelled.

Returns:

True only for the caller that wins the durable state transition.

async interrupt_task(record: QueuedTaskRecord, *, worker_id: str, reason: str = 'shutdown', max_interruptions: int | None = None) → QueuedTaskRecord | None[source]#

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:: The requeued or failed record, or None when the fence was lost.

async reset_task_identity(key: str) → bool[source]#

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:: True when a reservation was removed.

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

async execute_record(record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#

Execute a claimed queue record and persist the lifecycle result.

Parameters:

record – The claimed queue record to execute.
worker_id – 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.

Returns:

The updated queue record.

Raises:

asyncio.CancelledError – If task execution is cancelled.

async reconcile_external(limit: int | None = None) → int[source]#

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:: limit – When provided, examine at most this many external records across both halves. None reconciles every outstanding external record and repairs nothing.
Returns:: Number of records repaired or brought to a terminal queue status.

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

Recover stale running tasks and publish a worker summary event.

Parameters:

stale_after – Heartbeat age past which a running task is stale.
worker_id – Identity attached to published recovery events.
limit – When provided, recover at most this many records in one bounded batch. None preserves the unbounded worker behavior.

Returns:

Summary of recovered, failed, skipped, and handler-needed tasks.

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

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

Returns:: Records transitioned to expired.

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

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

Returns:: Records successfully transitioned to running.

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

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

async initialize_schedules() → list[QueuedTaskRecord][source]#

Create queue records for registered recurring schedules.

Returns:: The created or reused schedule records.

async publish_claim_lost(record: QueuedTaskRecord, *, phase: str, task_context: TaskExecutionContext | None = None, worker_id: str | None = None, expected_retry_count: int | None = None) → QueuedTaskRecord[source]#

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

Returns:: Current queue task record state.

Maintenance#

Bounded, backend-neutral queue maintenance.

The maintenance service runs a small, predictable amount of repair and retention work under token-fenced distributed coordination and a wall-clock time budget, then returns. It never starts a worker, executes due work, or loops to drain a backlog. Phases always run in the fixed order external-execution reconciliation, stale-running recovery, terminal-task retention, and durable-event retention.

litestar_queues.maintenance.MaintenancePhase#

One bounded maintenance phase, in fixed execution order.

alias of Literal[‘external’, ‘stale’, ‘terminal’, ‘events’]

litestar_queues.maintenance.MaintenancePhaseStatus#

Outcome of a single maintenance phase.

alias of Literal[‘completed’, ‘skipped’, ‘failed’, ‘partial’]

class litestar_queues.maintenance.QueueMaintenanceConfig(time_budget: float = 300.0, coordination_timeout: float = 360.0, external_limit: int = 100, stale_after: float | None = None, stale_limit: int = 100, terminal_retention: float | None = None, terminal_limit: int = 1000, event_retention_rules: tuple[QueueEventRetentionRule, ...] = (), event_limit: int = 1000)[source]#

Bases: object

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.

time_budget: float#: Maximum wall-clock duration of one maintenance run in seconds.

coordination_timeout: float#: Distributed ownership duration in seconds; must exceed time_budget.

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.

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.

Rules are evaluated in order and the first rule whose match matches a record decides that record’s retention age. A record matching no rule is never deleted.

event_limit: int#: Maximum expired task-event records deleted in one run.

__init__(time_budget: float = 300.0, coordination_timeout: float = 360.0, external_limit: int = 100, stale_after: float | None = None, stale_limit: int = 100, terminal_retention: float | None = None, terminal_limit: int = 1000, event_retention_rules: tuple[QueueEventRetentionRule, ...] = (), event_limit: int = 1000) → None#

class litestar_queues.maintenance.QueueMaintenancePhaseResult(phase: Literal['external', 'stale', 'terminal', 'events'], status: Literal['completed', 'skipped', 'failed', 'partial'], changed: int = 0, duration_ms: float = 0.0, error: str | None = None)[source]#

Bases: object

Result of one bounded maintenance phase.

phase: Literal['external', 'stale', 'terminal', 'events']#

status: Literal['completed', 'skipped', 'failed', 'partial']#

changed: int#

duration_ms: float#

error: str | None#

to_payload() → dict[str, object][source]#: Return a JSON-native mapping of this phase result.

__init__(phase: Literal['external', 'stale', 'terminal', 'events'], status: Literal['completed', 'skipped', 'failed', 'partial'], changed: int = 0, duration_ms: float = 0.0, error: str | None = None) → None#

class litestar_queues.maintenance.QueueMaintenanceService(service: QueueService, config: QueueMaintenanceConfig, *, monotonic: Callable[[], float] = <built-in function perf_counter>, utcnow: Callable[[], datetime] = <function _default_utcnow>)[source]#

Bases: object

Run bounded maintenance phases under token-fenced coordination and a time budget.

__init__(service: QueueService, config: QueueMaintenanceConfig, *, monotonic: Callable[[], float] = <built-in function perf_counter>, utcnow: Callable[[], datetime] = <function _default_utcnow>) → None[source]#

Initialize the maintenance service.

Parameters:

service – An opened queue service whose backend advertises supports_maintenance.
config – Bounded maintenance thresholds and limits.
monotonic – Injected monotonic clock for budget/duration accounting.
utcnow – Injected UTC clock used to compute stable retention cutoffs.

async run(phases: Collection[MaintenancePhase] | None = None) → QueueMaintenanceSummary[source]#

Claim maintenance ownership and run each selected phase once.

Parameters:: phases – Optional narrowing of the phases to run. Filtering only narrows configuration; it never enables a disabled retention threshold. None considers every phase in the fixed order.
Returns:: A summary whose outcome is already_running when ownership is denied, failed when any phase failed, partial when the budget skipped an enabled phase, else completed.
Raises:: QueueConfigurationError – If a requested phase name is unknown or the backend does not support distributed maintenance coordination.

class litestar_queues.maintenance.QueueMaintenanceSummary(outcome: ~typing.Literal['completed', 'failed', 'partial', 'already_running'], acquired: bool, duration_ms: float, phases: list[~litestar_queues.maintenance.QueueMaintenancePhaseResult] = <factory>)[source]#

Bases: object

Result of a whole maintenance run.

outcome: Literal['completed', 'failed', 'partial', 'already_running']#

acquired: bool#

duration_ms: float#

phases: list[QueueMaintenancePhaseResult]#

to_payload() → dict[str, object][source]#: Return a JSON-native mapping of the whole summary.

__init__(outcome: ~typing.Literal['completed', 'failed', 'partial', 'already_running'], acquired: bool, duration_ms: float, phases: list[~litestar_queues.maintenance.QueueMaintenancePhaseResult] = <factory>) → None#

Tasks#

class litestar_queues.task.RetryBackoff(initial_delay: float, multiplier: float = 1.0, max_delay: float | None = None)[source]#

Bases: object

Delay policy applied when a task consumes a retry.

initial_delay: float#

multiplier: float#

max_delay: float | None#

delay_for(retry_count: int) → float[source]#: Return the delay for a zero-based retry count.

as_metadata() → dict[str, float | None][source]#: Return the JSON-compatible persisted representation.

__init__(initial_delay: float, multiplier: float = 1.0, max_delay: float | None = None) → None#

Bases: object

Configuration for a recurring task schedule.

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(after: datetime | None = None, *, use_initial_delay: bool = False) → datetime[source]#

Calculate the next scheduled run time.

Returns:: The next run time in UTC.
Raises:: ValueError – If no interval or cron expression is configured.

as_metadata() → dict[str, Any][source]#: Return a JSON-compatible metadata representation.

copy_for_task(task_name: str) → ScheduleConfig[source]#: Return this normalized schedule for another task name.

__init__(task_name: str, cron: str | None = None, interval: timedelta | int | float | None = None, timezone: str = 'UTC', initial_delay: timedelta | int | float = 0, jitter: timedelta | int | float = 0) → None#

Bases: Generic[P, T]

Registered task wrapper with direct call and enqueue APIs.

__init__(func: TaskCallable[P, T], *, name: str, queue: str = 'default', priority: int = 0, retries: int = 0, retry_backoff: float | RetryBackoff | None = None, timeout: float | None = None, execution_backend: str | None = None, execution_profile: str | None = None, key: str | None = None, unique_by: TaskUniqueBy | None = None, unique_until: TaskUniqueUntil = 'terminal', run_after: float | timedelta | None = None, expires_in: float | timedelta | None = None, description: str | None = None, log_level: str | None = None, log_success: bool | None = None, requeue_on_stale: bool | None = None, requeue_on_shutdown: bool | None = None, on_stale_failure: StaleFailureHandler | None = None, sync_to_thread: bool | None = None, actor: QueueEventActor | Callable[[], QueueEventActor] | None = None) → None[source]#

property name: str#: Registered task name.

property queue: str#: Default queue name.

property priority: int#: Default priority.

property retries: int#: Maximum retry count.

property retry_backoff: RetryBackoff | None#: Retry delay policy, if configured.

property timeout: float | None#: Execution timeout.

property execution_backend: str | None#: Task-specific execution backend override.

property execution_profile: str | None#: Task-specific execution profile override.

property key: str | None#: Default deduplication key.

property unique_by: Literal['task', 'arguments'] | None#: Identity source when no explicit or configured key is supplied.

property unique_until: Literal['terminal', 'forever']#

terminal (default) or forever.

Type:: Identity lifetime

property signature: Signature#: Cached call signature of the wrapped callable, computed once per task.

property run_after: timedelta | None#: Relative delay for enqueue operations.

property expires_in: timedelta | None#: Relative not-started deadline for enqueue operations.

property description: str | None#: Task description metadata.

property log_level: str | None#: Task log level metadata.

property log_success: bool | None#: Whether successful completion logging should be quiet.

property requeue_on_stale: bool#: Whether stale running records should be requeued when retries remain.

property requeue_on_shutdown: bool | None#: Task-specific shutdown requeue override.

property on_stale_failure: Callable[[QueuedTaskRecord], object | Awaitable[object]] | None#: Callback invoked after this task reaches terminal stale failure.

property function: Callable[[P], T | Awaitable[T]]#: Wrapped callable.

property sync_to_thread: bool | None#

Whether a synchronous callable is offloaded to a worker thread.

None keeps the safe default and offloads; False opts in to running inline on the event loop.

property actor: QueueEventActor | Callable[[], QueueEventActor] | None#: Task-specific declared actor or actor resolver.

async execute_record(record: QueuedTaskRecord, *, task_context: TaskExecutionContext | None = None, extra_kwargs: Mapping[str, object] | None = None, sync_executor: Executor | None = None) → T[source]#

Execute this task for a queued record in worker context.

Returns:: The wrapped callable result.

metadata(values: dict[str, Any] | None = None) → dict[str, Any][source]#: Return enqueue metadata for this task.

using(*, queue: str | None = None, priority: int | None = None, retries: int | None = None, retry_backoff: float | RetryBackoff | None = None, timeout: float | None = None, execution_backend: str | None = None, execution_profile: str | None = None, key: str | None = None, unique_by: Literal['task', 'arguments'] | None = None, unique_until: Literal['terminal', 'forever'] | None = None, run_after: float | timedelta | None = None, expires_in: float | timedelta | None = None, description: str | None = None, log_level: str | None = None, log_success: bool | None = None, requeue_on_stale: bool | None = None, requeue_on_shutdown: bool | None = None, on_stale_failure: Callable[[QueuedTaskRecord], object | Awaitable[object]] | None = None, sync_to_thread: bool | None = None) → Task[P, T][source]#: Return a configured copy with enqueue overrides.

async enqueue(*args: ~typing.~P, **kwargs: ~typing.~P) → TaskResult[source]#

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

Returns:: A result handle for the queued record.

class litestar_queues.task.TaskResult(task_id: UUID, task_name: str, *, service: QueueService | None = None, record: QueuedTaskRecord | None = None)[source]#

Bases: object

Handle to a queued task result.

__init__(task_id: UUID, task_name: str, *, service: QueueService | None = None, record: QueuedTaskRecord | None = None) → None[source]#

property id: UUID#: Queue record ID.

property task_name: str#: Registered task name.

property status: TaskStatus | None#: Cached task status.

property result: Any#: Cached task result.

property error: str | None#: Cached task error.

property record: QueuedTaskRecord | None#: Cached queue record.

async refresh() → Self[source]#

Refresh this handle from its queue service.

Returns:: The refreshed result handle.
Raises:: RuntimeError – If the result has no associated service.

async wait(*, timeout: float | None = None, poll_interval: float = 0.1) → Self[source]#

Wait until the task reaches a terminal status.

Returns:

The completed result handle.

Raises:

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

litestar_queues.task.TaskUniqueBy#

the registered task name or the normalized call arguments.

Type:: Identity source for task uniqueness

alias of Literal[‘task’, ‘arguments’]

litestar_queues.task.TaskUniqueUntil#

released at terminal (the default) or retained by a reservation.

Type:: Identity lifetime for task uniqueness

alias of Literal[‘terminal’, ‘forever’]

litestar_queues.task.clear_task_registry() → None[source]#: Clear task and schedule registries.

litestar_queues.task.discover_tasks(package: str, subpackage: str = 'jobs', *, force_reload: bool = False) → tuple[str, ...][source]#

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:

package – Dotted package name to walk (e.g. "app.domain").
subpackage – Path segment that marks task modules. Any module whose dotted path (excluding the root) contains this segment is imported. Defaults to "jobs".
force_reload – Re-import modules already in sys.modules.

Returns:

Sorted, deduplicated tuple of task names registered after the walk.

Raises:

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

litestar_queues.task.get_default_service() → QueueService | None[source]#: Return the global default QueueService instance.

litestar_queues.task.get_scheduled_tasks() → dict[str, ScheduleConfig][source]#: Return the global scheduled task registry.

litestar_queues.task.get_task_registry() → dict[str, Task[Any, Any]][source]#: Return the global task registry.

litestar_queues.task.load_task_modules(modules: tuple[str, ...] | list[str], *, force_reload: bool = False) → int[source]#

Import configured task modules so decorators register tasks.

Returns:: Number of imported modules.

litestar_queues.task.set_default_service(service: QueueService | None) → None[source]#: Set the global default QueueService instance.

litestar_queues.task.task(func: Callable[[P], Awaitable[T]], /) → Task[P, T][source]#

litestar_queues.task.task(func: Callable[[P], T], /) → Task[P, T]

Returns:: A task wrapper when used bare, otherwise a decorator.
Raises:: ValueError – If both cron and interval are configured.

Models#

litestar_queues.models.TERMINAL_STATUSES: frozenset[Literal['pending', 'scheduled', 'running', 'completed', 'failed', 'cancelled', 'expired']] = frozenset({'cancelled', 'completed', 'expired', 'failed'})#: Statuses that represent finished queue records.

class litestar_queues.models.HeartbeatTouch(task_id: UUID, expected_retry_count: int | None, metadata_patch: dict[str, Any] | None = None)[source]#

Bases: object

A fenced heartbeat update request for one running task.

task_id: UUID#

expected_retry_count: int | None#

metadata_patch: dict[str, Any] | None#

__init__(task_id: UUID, expected_retry_count: int | None, metadata_patch: dict[str, Any] | None = None) → None#

class litestar_queues.models.HeartbeatTouchResult(touched_task_ids: set[~uuid.UUID] = <factory>, missed_task_ids: set[~uuid.UUID] = <factory>)[source]#

Bases: object

Backend-neutral result for a bulk heartbeat update.

touched_task_ids: set[UUID]#

missed_task_ids: set[UUID]#

__init__(touched_task_ids: set[~uuid.UUID] = <factory>, missed_task_ids: set[~uuid.UUID] = <factory>) → None#

class litestar_queues.models.QueueBackendCapabilities(supports_worker_wakeups: bool = False, wakeup_backend: str | None = None, wakeups_durable: bool = False, supports_completion_events: bool = False, supports_maintenance: bool = False)[source]#

Bases: object

Behavior advertised by a queue backend.

supports_worker_wakeups: bool#

wakeup_backend: str | None#

wakeups_durable: bool#

supports_completion_events: bool#

supports_maintenance: bool#

__init__(supports_worker_wakeups: bool = False, wakeup_backend: str | None = None, wakeups_durable: bool = False, supports_completion_events: bool = False, supports_maintenance: bool = False) → None#

class litestar_queues.models.QueueStatistics(pending: int = 0, scheduled: int = 0, running: int = 0, completed: int = 0, failed: int = 0, cancelled: int = 0, expired: int = 0)[source]#

Bases: object

Operational status counts for a queue backend.

pending: int#

scheduled: int#

running: int#

completed: int#

failed: int#

cancelled: int#

expired: int#

property total: int#: Total number of known queue records.

__init__(pending: int = 0, scheduled: int = 0, running: int = 0, completed: int = 0, failed: int = 0, cancelled: int = 0, expired: int = 0) → None#

class litestar_queues.models.QueuedTaskRecord(task_name: str, id: ~uuid.UUID = <factory>, args: tuple[~typing.Any, ...] = (), kwargs: dict[str, ~typing.Any] = <factory>, queue: str = 'default', execution_backend: str = 'local', execution_profile: str | None = None, execution_ref: str | None = None, worker_id: str | None = None, status: ~typing.Literal['pending', 'scheduled', 'running', 'completed', 'failed', 'cancelled', 'expired'] = 'pending', priority: int = 0, max_retries: int = 0, retry_count: int = 0, scheduled_at: ~datetime.datetime | None = None, expires_at: ~datetime.datetime | None = None, created_at: ~datetime.datetime = <factory>, queued_at: ~datetime.datetime = <factory>, started_at: ~datetime.datetime | None = None, completed_at: ~datetime.datetime | None = None, heartbeat_at: ~datetime.datetime | None = None, result: ~typing.Any | None = None, error: str | None = None, key: str | None = None, metadata: dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

Backend-neutral representation of a queued task.

task_name: str#

id: UUID#

args: tuple[Any, ...]#

kwargs: dict[str, Any]#

queue: str#

execution_backend: str#

execution_profile: str | None#

execution_ref: str | None#

worker_id: str | None#

status: Literal['pending', 'scheduled', 'running', 'completed', 'failed', 'cancelled', 'expired']#

priority: int#

max_retries: int#

retry_count: int#

scheduled_at: datetime | None#

expires_at: datetime | None#

created_at: datetime#

queued_at: datetime#

started_at: datetime | None#

completed_at: datetime | None#

heartbeat_at: datetime | None#

result: Any#

error: str | None#

key: str | None#

metadata: dict[str, Any]#

property is_terminal: bool#: Whether the record is in a terminal state.

property is_due: bool#: Whether the record is eligible to be claimed now.

__init__(task_name: str, id: ~uuid.UUID = <factory>, args: tuple[~typing.Any, ...] = (), kwargs: dict[str, ~typing.Any] = <factory>, queue: str = 'default', execution_backend: str = 'local', execution_profile: str | None = None, execution_ref: str | None = None, worker_id: str | None = None, status: ~typing.Literal['pending', 'scheduled', 'running', 'completed', 'failed', 'cancelled', 'expired'] = 'pending', priority: int = 0, max_retries: int = 0, retry_count: int = 0, scheduled_at: ~datetime.datetime | None = None, expires_at: ~datetime.datetime | None = None, created_at: ~datetime.datetime = <factory>, queued_at: ~datetime.datetime = <factory>, started_at: ~datetime.datetime | None = None, completed_at: ~datetime.datetime | None = None, heartbeat_at: ~datetime.datetime | None = None, result: ~typing.Any | None = None, error: str | None = None, key: str | None = None, metadata: dict[str, ~typing.Any] = <factory>) → None#

property is_expired: bool#: Whether the not-started deadline has passed.

class litestar_queues.models.StaleTaskRecoveryResult(requeued: int = 0, failed: int = 0, skipped: int = 0, handler_needed: int = 0, failed_task_ids: list[~uuid.UUID] = <factory>, handler_needed_task_ids: list[~uuid.UUID] = <factory>)[source]#

Bases: object

Summary of stale running task recovery.

requeued: int#

failed: int#

skipped: int#

handler_needed: int#

failed_task_ids: list[UUID]#

handler_needed_task_ids: list[UUID]#

to_payload() → dict[str, int][source]#: Return a JSON-compatible event payload.

__init__(requeued: int = 0, failed: int = 0, skipped: int = 0, handler_needed: int = 0, failed_task_ids: list[~uuid.UUID] = <factory>, handler_needed_task_ids: list[~uuid.UUID] = <factory>) → None#

class litestar_queues.models.TaskRequest(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)[source]#

Bases: object

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.

task_name: str#

args: tuple[Any, ...]#

kwargs: dict[str, Any] | None#

queue: str#

priority: int#

max_retries: int#

scheduled_at: datetime | None#

expires_at: datetime | None#

key: str | None#

execution_backend: str#

execution_profile: str | None#

metadata: dict[str, Any] | None#

__init__(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) → None#

class litestar_queues.models.TaskReservation(key: str, task_id: UUID, task_name: str, created_at: datetime)[source]#

Bases: object

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.

key: str#

task_id: UUID#

task_name: str#

created_at: datetime#

__init__(key: str, task_id: UUID, task_name: str, created_at: datetime) → None#

litestar_queues.models.TaskStatus#

Queue task lifecycle states.

alias of Literal[‘pending’, ‘scheduled’, ‘running’, ‘completed’, ‘failed’, ‘cancelled’, ‘expired’]

Worker#

Everything that runs queue work, grouped by who owns the process.

worker is the claim/execute loop itself, the only public name here.
runtime orchestrates one worker start-to-stop without knowing how it was launched. Both the CLI and the server child use it.
supervisor owns the fresh worker child that a litestar run invocation starts, plus that child’s process entry point.
invocation publishes and verifies the marker that tells a process whether its invocation already owns a worker.
heartbeat keeps claims alive while work runs.

Submodules are imported lazily so importing Worker does not pull in multiprocessing or the Litestar CLI.

class litestar_queues.worker.Worker(service: QueueService, config: WorkerConfig | None = None)[source]#

Bases: object

Local in-process queue worker.

__init__(service: QueueService, config: WorkerConfig | None = None) → None[source]#

Initialize the worker.

Parameters:

service – Queue service used to reach the configured backends.
config – Worker runtime configuration; None uses defaults.

property is_running: bool#: Whether the worker loop is active.

property worker_id: str#: Worker identity used for events and logs.

async wait_started() → None[source]#: Wait until heartbeat startup succeeds or propagate its failure.

async start() → None[source]#: Run the worker loop until stopped or cancelled.

async stop(*, force: bool = False) → bool[source]#

Stop the worker loop and drain or cancel in-flight work.

Returns:: True when graceful drain escalated to cancellation.

async run_once() → int[source]#

Process one batch of due tasks.

Returns:: Number of claimed task records.
Raises:: QueueConfigurationError – If the execution backend schedules its own delivery, leaving a worker nothing it may safely claim.

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

Reconcile externally dispatched records by delegating to the service.

The reconciliation contract (state transitions, unknown-backend skipping, and metrics) lives on QueueService.reconcile_external(); the worker keeps only the periodic cadence and fleet lock.

Returns:: Number of records that reached a terminal queue status.

Plugin#

class litestar_queues.plugin.QueuePlugin(config: QueueConfig | None = None)[source]#

Bases: InitPlugin, CLIPlugin

Litestar plugin for queue service dependency registration and lifecycle.

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

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

property config: QueueConfig#: Plugin configuration.

get_service(state: State | None = None) → QueueService[source]#: Return a QueueService for this plugin.

create_worker_service() → QueueService[source]#: Create a fresh service that owns its process-local event resources.

on_app_init(app_config: AppConfig) → AppConfig[source]#

Returns:: The updated application configuration.

on_cli_init(cli: ClickGroup) → None[source]#

Attach the queues subcommand group to the Litestar CLI.

Parameters:: cli – The root click.Group of the Litestar CLI.

server_lifespan(app: Litestar) → Generator[None][source]#

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:: None – with the invocation marker, storage, and worker child active.

Observability#

litestar_queues.observability.DEFAULT_DURATION_BUCKETS = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 300.0, 600.0, 1800.0)#

Buckets spanning sub-millisecond enqueues through half-hour task executions.

The prometheus_client default tops out at ten seconds, which sends every real task duration into the +Inf bucket.

class litestar_queues.observability.ObservabilityConfig(enable_otel: bool | None = None, enable_prometheus: bool | None = None, enable_sqlcommenter: bool | None = None, tracer_name: str | None = None, meter_name: str | None = None, tracer_provider: Any | None = None, meter_provider: Any | None = None, prometheus_registry: Any | None = None, metric_prefix: str | None = None, duration_buckets: tuple[float, ...] = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 300.0, 600.0, 1800.0))[source]#

Bases: object

Configuration for optional queue-domain observability.

enable_otel: bool | None#: OpenTelemetry policy; None follows the active Litestar telemetry plugin.

enable_prometheus: bool | None#: Prometheus policy; None follows the app’s Litestar Prometheus middleware.

enable_sqlcommenter: bool | None#: SQLCommenter policy; None follows resolved queue telemetry.

tracer_name: str | None#: Instrumentation name; None uses QueueConfig.namespace.

meter_name: str | None#: Instrumentation name; None uses QueueConfig.namespace.

tracer_provider: Any | None#: Explicit OpenTelemetry tracer provider; None uses the global provider.

meter_provider: Any | None#: Explicit OpenTelemetry meter provider; None uses the global provider.

prometheus_registry: Any | None#: Explicit Prometheus registry; None uses the client default registry.

metric_prefix: str | None#: Prometheus prefix; None uses QueueConfig.namespace.

duration_buckets: tuple[float, ...]#: Prometheus histogram buckets, in seconds, for queue duration metrics.

should_enable_otel(app: Litestar | None = None) → bool[source]#

Return whether OpenTelemetry should be enabled.

Returns:: Whether OpenTelemetry tracing and metrics should be enabled.

should_enable_prometheus(app: Litestar | None = None) → bool[source]#

Return whether Prometheus metrics should be enabled.

Returns:: Whether Prometheus metrics should be enabled.

should_enable_sqlcommenter(app: Litestar | None = None) → bool[source]#

Return whether SQLCommenter attribution should be enabled.

Returns:: Whether queue-owned SQLCommenter attribution should be enabled.

resolve_prometheus_registry() → Any[source]#

Return the registry queue collectors are registered with.

Returns:: The configured registry, or the prometheus_client default registry, which is what Litestar’s PrometheusController scrapes.

__init__(enable_otel: bool | None = None, enable_prometheus: bool | None = None, enable_sqlcommenter: bool | None = None, tracer_name: str | None = None, meter_name: str | None = None, tracer_provider: Any | None = None, meter_provider: Any | None = None, prometheus_registry: Any | None = None, metric_prefix: str | None = None, duration_buckets: tuple[float, ...] = (0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0, 300.0, 600.0, 1800.0)) → None#

class litestar_queues.observability.QueueObservabilityRuntime(config: ObservabilityConfig | None, *, app: Litestar | None = None, namespace: QueueNamespace | str | None = None)[source]#

Bases: object

Runtime helper for queue-domain spans and metrics.

__init__(config: ObservabilityConfig | None, *, app: Litestar | None = None, namespace: QueueNamespace | str | None = None) → None[source]#

enabled#

property sqlcommenter_enabled: bool#: Whether backends should attach SQLCommenter attribution to statements.

get_tracer() → Any[source]#

Return the configured tracer.

Returns:: The configured OpenTelemetry tracer.

get_meter() → Any[source]#

Return the configured meter.

Returns:: The configured OpenTelemetry meter.

start_span(name: str, *, kind: str, attributes: Mapping[str, object], parent: object | None = None) → Any | None[source]#

Start a queue span and make it the current span.

The span must be current for two reasons: inject_trace_context serialises the current context, and any instrumentation running inside the span – database drivers, HTTP clients, log correlation – resolves its parent from the current context.

Returns:: The started span handle, or None when tracing is disabled.

end_span(span: Any | None) → None[source]#: End a span and restore the previous current span.

record_exception(span: Any | None, exc: BaseException) → None[source]#: Record an exception on a span and mark the span failed.

set_status_error(span: Any | None, description: str) → None[source]#: Mark a span as failed when no exception reached this frame.

set_attribute(span: Any | None, key: str, value: object) → None[source]#: Set a span attribute if one was created.

inject_trace_context(metadata: dict[str, Any]) → None[source]#: Inject current W3C trace context into queue metadata.

extract_trace_context(metadata: Mapping[str, Any]) → object | None[source]#

Extract a parent trace context from queue metadata.

Returns:: Extracted trace context, or None when unavailable.

record_counter(name: str, value: int = 1, *, attributes: Mapping[str, str]) → None[source]#: Record a counter value for enabled metrics sinks.

record_gauge_delta(name: str, delta: int = 1, *, attributes: Mapping[str, str]) → None[source]#: Record a gauge delta for enabled metrics sinks.

record_duration(name: str, seconds: float, *, attributes: Mapping[str, str]) → None[source]#: Record a duration for enabled metrics sinks.

record_histogram(name: str, value: float, *, unit: str, attributes: Mapping[str, str]) → None[source]#: Record a non-duration histogram sample for enabled metric sinks.

class litestar_queues.observability.QueueObservabilityRuntimeProtocol(*args, **kwargs)[source]#

Bases: Protocol

Protocol for queue observability runtimes used by services and workers.

enabled: bool#

start_span(name: str, *, kind: str, attributes: Mapping[str, object], parent: object | None = None) → Any | None[source]#

Start a queue span and make it the current span.

Returns:: The started span handle, or None when tracing is disabled.

end_span(span: Any | None) → None[source]#: End a span and restore the previous current span.

record_exception(span: Any | None, exc: BaseException) → None[source]#: Record an exception on a span and mark it failed.

set_status_error(span: Any | None, description: str) → None[source]#: Mark a span as failed without an exception.

set_attribute(span: Any | None, key: str, value: object) → None[source]#: Set a span attribute.

inject_trace_context(metadata: dict[str, Any]) → None[source]#: Inject trace context into queue metadata.

extract_trace_context(metadata: Mapping[str, Any]) → object | None[source]#

Extract trace context from queue metadata.

Returns:: Extracted trace context, or None when unavailable.

record_counter(name: str, value: int = 1, *, attributes: Mapping[str, str]) → None[source]#: Record a counter sample.

record_gauge_delta(name: str, delta: int = 1, *, attributes: Mapping[str, str]) → None[source]#: Record a gauge delta sample.

record_duration(name: str, seconds: float, *, attributes: Mapping[str, str]) → None[source]#: Record a duration sample.

record_histogram(name: str, value: float, *, unit: str, attributes: Mapping[str, str]) → None[source]#: Record a value histogram sample.

__init__(*args, **kwargs)#

litestar_queues.observability.bind_correlation_id(metadata: Mapping[str, Any]) → tuple[Any, bool][source]#

Rebind the enqueueing request’s correlation ID for task execution.

Returns:: The previously active correlation ID, and whether it must be restored.

litestar_queues.observability.capture_correlation_id(metadata: dict[str, Any]) → None[source]#: Store the currently active correlation ID on a queued record.

litestar_queues.observability.create_observability_runtime(config: ObservabilityConfig | None, *, app: Litestar | None = None, namespace: QueueNamespace | str | None = None) → QueueObservabilityRuntime[source]#

Create the queue observability runtime for a service.

Returns:: Queue observability runtime instance.

litestar_queues.observability.reset_correlation_id(state: tuple[Any, bool]) → None[source]#: Restore the correlation ID that was active before task execution.

Typing#

Public typing helpers for optional observability support.

The supported import location for litestar_queues._typing. Each package publishes its own facade over its own private module – this one does not re-export a nested package’s types, so event backend protocols live in litestar_queues.events.typing and adapter protocols beside their adapter.

litestar_queues.typing.OtelMeter#: alias of Meter

litestar_queues.typing.OtelSpan#: alias of Span

litestar_queues.typing.OtelSpanKind#: alias of SpanKind

litestar_queues.typing.OtelStatus#: alias of Status

litestar_queues.typing.OtelStatusCode#: alias of StatusCode

litestar_queues.typing.OtelTracer#: alias of Tracer

litestar_queues.typing.PrometheusCounter#: alias of Counter

litestar_queues.typing.PrometheusGauge#: alias of Gauge

litestar_queues.typing.PrometheusHistogram#: alias of Histogram

litestar_queues.typing.prometheus_default_registry() → Any[source]#

Return the default Prometheus registry.

Returns:: The process-wide prometheus_client registry that Litestar’s PrometheusController scrapes.

litestar_queues.typing.sqlspec_correlation_context() → Any | None[source]#

Return SQLSpec’s CorrelationContext.

Resolved lazily rather than at import time, because importing SQLSpec is not free and most consumers of this module never need it.

Returns:: SQLSpec’s CorrelationContext class, or None when SQLSpec is absent.

Exceptions#

exception litestar_queues.exceptions.JobCancelledError[source]#

Bases: QueueError

Raised by a task to cooperatively mark itself cancelled.

exception litestar_queues.exceptions.MissingDependencyError(package: str, install_package: str | None = None)[source]#

Bases: QueueError, ImportError

Raised when a required optional dependency is not installed.

__init__(package: str, install_package: str | None = None) → None[source]#

Initialize missing dependency error.

Parameters:

package – The missing import package.
install_package – Optional package or extra to install.

exception litestar_queues.exceptions.NonRetryableError[source]#

Bases: QueueError

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

exception litestar_queues.exceptions.QueueConfigurationError[source]#

Bases: QueueError

Raised when queue backend configuration is invalid.

exception litestar_queues.exceptions.QueueDispatchError(message: str, *, task_id: UUID, committed: bool)[source]#

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.

__init__(message: str, *, task_id: UUID, committed: bool) → None[source]#

Initialize dispatch error.

Parameters:

message – Human-readable description of the dispatch failure.
task_id – Identifier of the record that failed to dispatch.
committed – Whether the record is durably persisted.

exception litestar_queues.exceptions.QueueError[source]#

Bases: Exception

Base exception for litestar-queues errors.

exception litestar_queues.exceptions.QueueEventBufferFull[source]#

Bases: QueueError

Raised when queue event buffering cannot accept another event.

exception litestar_queues.exceptions.QueueWarning[source]#

Bases: UserWarning

Base class for litestar-queues warnings.

exception litestar_queues.exceptions.TaskIdentityError[source]#

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.

exception litestar_queues.exceptions.TaskIdentityTooLargeError(*, actual_bytes: int, max_bytes: int)[source]#

Bases: QueueError

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

__init__(*, actual_bytes: int, max_bytes: int) → None[source]#

Initialize the payload-size error.

Parameters:

actual_bytes – Measured size of the canonical identity payload.
max_bytes – Configured QueueConfig.max_argument_identity_bytes limit.

litestar_queues.exceptions.job_cancelled(message: str = 'Task cancelled') → None[source]#

Raise a cooperative task cancellation.

Raises:: JobCancelledError – Always raised with the provided message.

litestar_queues.exceptions.non_retryable(message: str) → None[source]#

Raise a non-retryable task failure.

Raises:: NonRetryableError – Always raised with the provided message.