Execution Backends#

Registry#

Execution backends are registered through litestar_queues.execution.

litestar_queues.execution.get_execution_backend(backend: ExecutionBackendConfig = 'immediate', config: QueueConfig | None = None) → BaseExecutionBackend[source]#

Get an instantiated execution backend.

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

litestar_queues.execution.get_execution_backend_class(backend_path: str) → type[BaseExecutionBackend][source]#

Get an execution backend class by short name or import path.

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

litestar_queues.execution.execution_backend(name: str) → Callable[[type[BaseExecutionBackend]], type[BaseExecutionBackend]][source]#

Decorator to register an execution backend class with a short name.

Returns:: A decorator that registers the backend class.

litestar_queues.execution.list_execution_backends() → list[str][source]#: Return registered execution backend names.

Base Backend#

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

Bases: BaseExecutionBackend

Base for external backends that continuously receive broker deliveries.

async run_consumer(service: QueueService, *, max_concurrency: int, drain_timeout: float) → None[source]#: Receive and execute deliveries until cancelled.

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

Bases: object

Base class for queue execution backends.

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

config#

property is_external: bool#: Whether this backend dispatches records to another process.

property schedules_on_enqueue: bool#

Whether the backend schedules a persisted record without a Worker.

Deliberately separate from is_external. Cloud Run Jobs and the broker backends are external but still rely on a worker loop noticing a pending record; only a managed transport that accepts the record itself answers true here.

property max_schedule_horizon: timedelta | None#

How far ahead this backend will hold a scheduled delivery.

None means unbounded, which is every backend that keeps due records in the queue store. A managed transport that takes ownership of the record has its own ceiling, and a schedule past it is not a call that fails once – it is a recurrence that can never run.

async schedule(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Schedule one already-persisted record for external delivery.

Returns:: The external delivery reference, if one was created.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#

Recreate deliveries this backend owns that its transport no longer holds.

A no-op for every backend whose records are found by polling: nothing can go missing from a store the worker reads directly. Only a managed transport that took ownership of the record can lose it silently.

Parameters:

service – The queue service whose records to repair.
limit – Ceiling on how many records one pass may examine. Bounded maintenance is the only caller, and it always passes a positive budget it needs back.

Returns:

An empty result.

async open() → bool[source]#

Open execution resources.

Returns:: True when resources are ready.

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

async execute(service: QueueService, record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#: Execute a queue record.

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Dispatch a queue record to an external executor.

Returns:: The external execution reference, if one was created.

async reconcile(service: QueueService, record: QueuedTaskRecord) → QueuedTaskRecord | None[source]#

Reconcile an externally running queue record.

Returns:: The updated record when reconciliation changes state.

async cancel_execution(service: QueueService, record: QueuedTaskRecord) → ExecutionCancelResult[source]#

Cancel the provider resource backing one external attempt.

Called before any durable transition to cancelled for a record whose execution_ref names a live provider resource. Implementations must contain their client’s exceptions and answer with a result: a not-found resource is already_cancelled, anything transient or refused is retryable. Raising is reserved for programming errors.

Returns:: unsupported for a transport with no cancellation control plane.

class litestar_queues.execution.base.DispatchRepairResult(examined: int = 0, changed: int = 0)[source]#

Bases: object

Outcome of one bounded delivery-repair pass.

examined is how much of the caller’s budget the pass consumed, whether or not a candidate needed anything done, so the caller can spend what is left on its other work.

examined: int#

changed: int#

__init__(examined: int = 0, changed: int = 0) → None#

class litestar_queues.execution.base.ExecutionCancelResult(status: Literal['accepted', 'already_cancelled', 'retryable', 'unsupported'], detail: str | None = None)[source]#

Bases: object

Outcome of one provider-level cancellation attempt.

detail carries the provider’s own words for the log line and the lifecycle event; it is never parsed.

status: Literal['accepted', 'already_cancelled', 'retryable', 'unsupported']#

detail: str | None#

property permits_durable_cancel: bool#

Whether this outcome may win the durable transition to cancelled.

Only a control plane that exists and refused blocks the write. A transport with no control plane cannot strand anything, because the dispatch-by-id claim fence makes an in-flight delivery a no-op.

Returns:: True unless the provider refused a cancellation it could have made.

classmethod accepted(detail: str | None = None) → Self[source]#

Return an accepted cancellation result.

Returns:: A result whose status is accepted.

classmethod already_cancelled(detail: str | None = None) → Self[source]#

Return an idempotent already-cancelled result.

Returns:: A result whose status is already_cancelled.

classmethod retryable(detail: str | None = None) → Self[source]#

Return a transient or rejected cancellation result.

Returns:: A result whose status is retryable.

classmethod unsupported(detail: str | None = None) → Self[source]#

Return a result for a transport with no cancellation control plane.

Returns:: A result whose status is unsupported.

__init__(status: Literal['accepted', 'already_cancelled', 'retryable', 'unsupported'], detail: str | None = None) → None#

Immediate#

class litestar_queues.execution.immediate.ImmediateExecutionBackend(config: QueueConfig | None = None)[source]#

Bases: BaseExecutionBackend

Execution backend that runs records inline.

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

Execute a task immediately in the current event loop.

Returns:: The updated queue record.

Local#

class litestar_queues.execution.local.LocalExecutionBackend(config: QueueConfig | None = None)[source]#

Bases: BaseExecutionBackend

Execution backend for in-process workers.

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

Execute one claimed task in a local worker.

Returns:: The updated queue record.

Cloud Run#

class litestar_queues.execution.cloudrun.config.CloudRunExecutionConfig(project_id: str, region: str = 'us-central1', job_name: str | None = None, profiles: dict[str, str] = <factory>, timeout: int = 300, env_prefix: str | None = None, extra_env: dict[str, str] = <factory>, fallback_execution_backend: str | None = None)[source]#

Bases: object

Configuration for Cloud Run Jobs execution.

backend_name: ClassVar[str] = 'cloudrun'#

project_id: str#: Google Cloud project containing the target jobs.

region: str#: Google Cloud region containing the target jobs.

job_name: str | None#: Default Cloud Run Job name; None requires a matching profile.

profiles: dict[str, str]#: Execution-profile names mapped to Cloud Run Job names.

timeout: int#: Cloud Run API operation timeout in seconds.

env_prefix: str | None#: Explicit environment prefix; None derives it from QueueConfig.namespace.

__init__(project_id: str, region: str = 'us-central1', job_name: str | None = None, profiles: dict[str, str] = <factory>, timeout: int = 300, env_prefix: str | None = None, extra_env: dict[str, str] = <factory>, fallback_execution_backend: str | None = None) → None#

extra_env: dict[str, str]#: Additional environment variables passed to every Cloud Run execution.

fallback_execution_backend: str | None#: Backend used after dispatch failure; None propagates the failure.

resolve_job_name(profile: str | None = None) → str[source]#

Return the Cloud Run Job name for a profile.

Returns:: The resolved Cloud Run Job name.
Raises:: QueueConfigurationError – If no job name can be resolved.

env_name(suffix: str, *, namespace: QueueNamespace | None = None) → str[source]#: Return an environment variable name using the configured prefix.

class litestar_queues.execution.cloudrun.backend.CloudRunExecutionBackend(config: QueueConfig | None = None, *, execution_config: CloudRunExecutionConfig | None = None, jobs_client: CloudRunJobsClient | None = None, executions_client: CloudRunExecutionsClient | None = None)[source]#

Bases: BaseExecutionBackend

Execution backend that dispatches queued records to Cloud Run Jobs.

__init__(config: QueueConfig | None = None, *, execution_config: CloudRunExecutionConfig | None = None, jobs_client: CloudRunJobsClient | None = None, executions_client: CloudRunExecutionsClient | None = None) → None[source]#: Initialize the execution backend.

jobs_client#

executions_client#

property is_external: bool#: Whether this backend dispatches records to another process.

property execution_config: CloudRunExecutionConfig#: Resolved Cloud Run execution config.

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

Dispatch a record and return its persisted state.

The worker_id argument is accepted for protocol parity but not forwarded: external dispatch does not run service.execute_record locally, so the remote runner is responsible for its own worker identity binding.

Returns:: The persisted queue record after dispatch.

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Dispatch a queue record to Cloud Run Jobs.

Returns:: The Cloud Run execution reference, if dispatch succeeds.

async cancel_execution(service: QueueService, record: QueuedTaskRecord) → ExecutionCancelResult[source]#

Cancel the Cloud Run execution backing this record’s attempt.

Returns:: The provider’s answer, mapped to the shared cancellation contract.

async reconcile(service: QueueService, record: QueuedTaskRecord) → QueuedTaskRecord | None[source]#

Reconcile a Cloud Run execution with the queue record.

Returns:: The terminal queue record when reconciliation completed it.

async check_execution_status(execution_ref: str) → CloudRunExecutionStatus[source]#

Return Cloud Run execution status.

Transient API failures are treated as still running so reconciliation does not create false terminal queue states.

build_run_job_request(service: QueueService, record: QueuedTaskRecord) → dict[str, Any][source]#

Build the Cloud Run Jobs API request for a queue record.

Returns:: Cloud Run Jobs API request data.

build_task_env(record: QueuedTaskRecord) → dict[str, str][source]#

Build the single-value task environment for a Cloud Run task process.

The record travels as its id in one prefix-aware environment variable (QUEUES_TASK_ID); the consumer re-fetches the live record by that id. Adopter extra_env values are merged in.

Returns:: Environment variables for the Cloud Run task process.

class litestar_queues.execution.cloudrun.backend.CloudRunExecutionStatus(succeeded: bool = False, failed: bool = False, cancelled: bool = False, running: bool = True, error: str | None = None)[source]#

Bases: object

Backend-neutral status for a Cloud Run execution.

succeeded: bool#

failed: bool#

cancelled: bool#

running: bool#

error: str | None#

__init__(succeeded: bool = False, failed: bool = False, cancelled: bool = False, running: bool = True, error: str | None = None) → None#

Cloud Tasks#

litestar_queues.execution.cloudtasks.config.CLOUD_TASKS_BACKEND_NAME = 'cloudtasks'#: Registry name of this execution backend, and its value on every metric label.

litestar_queues.execution.cloudtasks.config.CLOUD_TASKS_MAX_SCHEDULE_HORIZON = datetime.timedelta(days=30)#: Google’s fixed ceiling on how far ahead a Cloud Tasks task may be scheduled.

litestar_queues.execution.cloudtasks.config.CLOUD_TASKS_PROTOCOL_VERSION = 1#

Version stamped on every delivery body and required by the consumer route.

Lives here rather than beside either side of the wire so the producer and the route cannot drift apart while both still look correct on their own.

class litestar_queues.execution.cloudtasks.config.CloudTasksExecutionConfig(project_id: str, location: str, queue_id: str, service_url: str, service_account_email: str, audience: str | None = None, route_path: str = <object object>, delivery_name_prefix: str | None = None, dispatch_deadline: int = 1800, response_margin: float = 30.0, default_task_timeout: float = 1740.0, api_timeout: float = 10.0, trust_platform_auth: bool = False, guards: tuple[Guard, ...] = ())[source]#

Bases: object

Configuration for Google Cloud Tasks managed dispatch.

Cloud Tasks delivers each persisted record over HTTP to a private consumer service, so this carries both the queue coordinates and the delivery target.

backend_name: ClassVar[str] = 'cloudtasks'#

project_id: str#: Google Cloud project owning the Cloud Tasks queue.

location: str#: Cloud Tasks queue location, such as us-central1.

queue_id: str#: Cloud Tasks queue that receives every dispatched record.

service_url: str#: HTTPS origin of the private consumer service.

service_account_email: str#: Service account Cloud Tasks mints the OIDC delivery token for.

audience: str | None#: OIDC audience; None resolves to the service_url origin.

route_path: str#: Path the consumer route is mounted on.

delivery_name_prefix: str | None#: Cloud Tasks task-name prefix; None derives it from QueueConfig.namespace.

dispatch_deadline: int#: Seconds Cloud Tasks waits for a delivery response before abandoning it.

response_margin: float#: Seconds reserved for the consumer to answer inside the dispatch deadline.

__init__(project_id: str, location: str, queue_id: str, service_url: str, service_account_email: str, audience: str | None = None, route_path: str = <object object>, delivery_name_prefix: str | None = None, dispatch_deadline: int = 1800, response_margin: float = 30.0, default_task_timeout: float = 1740.0, api_timeout: float = 10.0, trust_platform_auth: bool = False, guards: tuple[Guard, ...] = ()) → None#

default_task_timeout: float#: Timeout applied to records that declare none of their own.

api_timeout: float#: Timeout for a single Cloud Tasks API call.

trust_platform_auth: bool#: Whether Cloud Run IAM alone is accepted as protection for the route.

guards: tuple[Guard, ...]#: Application guards applied to the consumer route.

property target_url: str#: Absolute URL Cloud Tasks posts each record to.

resolve(namespace: QueueNamespace | None = None) → CloudTasksExecutionConfig[source]#: Resolve namespace-owned defaults without mutating this reusable config.

property queue_path: str#: Fully qualified Cloud Tasks queue resource name.

Cloud Tasks execution backend.

Constructing the backend never builds a Google client: the client is resolved on first use so an installation without the cloud-tasks extra can still import, configure, and validate this backend.

Scheduling only ever runs against an already-durable record, and the delivery’s resource name is persisted before the create call so an ambiguous response still leaves a handle to look the delivery up by.

class litestar_queues.execution.cloudtasks.backend.CloudTasksExecutionBackend(config: QueueConfig | None = None, *, execution_config: CloudTasksExecutionConfig | None = None, client: CloudTasksClient | None = None)[source]#

Bases: BaseExecutionBackend

Execution backend that hands persisted records to Google Cloud Tasks.

__init__(config: QueueConfig | None = None, *, execution_config: CloudTasksExecutionConfig | None = None, client: CloudTasksClient | None = None) → None[source]#: Initialize the Cloud Tasks execution backend.

client#

property is_external: bool#: Whether this backend dispatches records to another process.

property schedules_on_enqueue: bool#: Whether the backend schedules a persisted record without a Worker.

property max_schedule_horizon: timedelta#: How far ahead Cloud Tasks will hold a scheduled delivery.

property execution_config: CloudTasksExecutionConfig#: Resolved Cloud Tasks execution config.

async schedule(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Create the Cloud Tasks delivery for an already-persisted record.

The record is read back rather than trusted: between the caller’s copy and this call it may have been cancelled, expired, or removed, and a delivery for an id storage no longer holds would retry against the consumer until it expired.

Returns:: The delivery’s full resource name, or None when the record is no longer eligible.
Raises:: Exception – Whatever prevented the delivery from being created, after the failure has been marked on the span. The record is committed, so the caller must not retry the enqueue.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#

Recreate deliveries Cloud Tasks no longer holds for still-active records.

A delivery can go missing while its record stays active: a create call that failed after Google had already accepted it, an operator purging the queue, a retention window closing before the schedule time arrived. Nothing polls these records, so without this pass they wait forever.

Every candidate is attempted at most once per pass, so a queue whose target is broken cannot spin inside one maintenance window.

Parameters:

service – The queue service whose records to repair.
limit – Ceiling on how many records this pass may examine.

Returns:

How many records the pass looked at, and how many it re-delivered.

async close() → None[source]#

Release a client this backend created.

An injected client belongs to whoever built it and is left untouched.

async cancel_execution(service: QueueService, record: QueuedTaskRecord) → ExecutionCancelResult[source]#

Delete the Cloud Tasks delivery holding this record’s attempt.

Cloud Tasks holds a future delivery rather than a running process, so cancellation is a delete: after it, nothing will be delivered. A delivery that is already in flight has been claimed by the consumer, and the delete answers not-found – idempotent success, with the running body left to cooperative cancellation.

Returns:: The provider’s answer, mapped to the shared cancellation contract.

The private HTTP route Cloud Tasks delivers each record to.

Cloud Tasks treats any non-2xx response as “deliver this again”. That is why this route answers so few statuses: every outcome the queue reached durably is acknowledged, whether the task succeeded, failed for the last time, was cancelled, or was already owned by someone else. None of those change if the same delivery arrives a second time.

Only two things earn a retryable answer – the queue could not be reached, and the retry this request just scheduled never made it to Google – because in both cases the redelivery is what repairs the record.

A request that cannot be authenticated or parsed is never acknowledged. That is a deployment fault, and letting Cloud Tasks retry it is what makes it visible.

class litestar_queues.execution.cloudtasks.routes.CloudTasksDelivery(version: int, task_id: UUID)[source]#

Bases: Struct

Everything that crosses the transport for one delivery.

Rejecting unknown fields is the point rather than strictness for its own sake: the record in storage is authoritative, so anything a caller adds here would either be ignored or, worse, believed.

litestar_queues.execution.cloudtasks.routes.build_cloud_tasks_route(queue_config: QueueConfig) → HTTPRouteHandler[source]#

Build the delivery route for a queue configured for Cloud Tasks.

Returns:: The route handler Cloud Tasks posts each record to.

Amazon SQS#

class litestar_queues.execution.sqs.config.SqsExecutionConfig(queue_url: str, region_name: str | None = None, endpoint_url: str | None = None, fifo: bool = False, message_group_id: str | None = None, wait_time_seconds: int = 20, receive_batch_size: int = 10, visibility_timeout: int = 60, visibility_extension_interval: int = 30, dispatch_stale_after: int = 60, api_timeout: float = 30)[source]#

Bases: object

Amazon SQS execution-dispatch configuration.

backend_name: ClassVar[str] = 'sqs'#

queue_url: str#: Absolute URL of the target SQS queue.

region_name: str | None#: Optional AWS region override; the normal SDK chain supplies the default.

endpoint_url: str | None#: Optional endpoint override, primarily for LocalStack.

fifo: bool#: Whether to emit FIFO-only group and deduplication fields.

message_group_id: str | None#: Explicit FIFO group; when omitted a stable queue-name group is derived.

wait_time_seconds: int#: SQS long-poll duration.

receive_batch_size: int#: Maximum number of messages requested per receive.

__init__(queue_url: str, region_name: str | None = None, endpoint_url: str | None = None, fifo: bool = False, message_group_id: str | None = None, wait_time_seconds: int = 20, receive_batch_size: int = 10, visibility_timeout: int = 60, visibility_extension_interval: int = 30, dispatch_stale_after: int = 60, api_timeout: float = 30) → None#

visibility_timeout: int#: Initial SQS delivery visibility, independent from queue heartbeat leases.

visibility_extension_interval: int#: Courtesy visibility-extension cadence during local execution.

dispatch_stale_after: int#: Age after which an owned attempt can be atomically rotated and republished.

api_timeout: float#: Timeout applied to individual SQS API operations.

class litestar_queues.execution.sqs.backend.SqsExecutionBackend(config: QueueConfig | None = None, *, execution_config: SqsExecutionConfig | None = None, client: Any | None = None)[source]#

Bases: BaseConsumerExecutionBackend

Dispatch bare task identifiers through Amazon SQS.

__init__(config: QueueConfig | None = None, *, execution_config: SqsExecutionConfig | None = None, client: Any | None = None) → None[source]#: Initialize the execution backend.

property is_external: bool#: Whether this backend dispatches records to another process.

property execution_config: SqsExecutionConfig#

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Dispatch a queue record to an external executor.

Returns:: The external execution reference, if one was created.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#

Recreate deliveries this backend owns that its transport no longer holds.

Parameters:

service – The queue service whose records to repair.
limit – Ceiling on how many records one pass may examine. Bounded maintenance is the only caller, and it always passes a positive budget it needs back.

Returns:

An empty result.

async execute(service: QueueService, record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#: Execute a queue record.

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

async run_consumer(service: QueueService, *, max_concurrency: int, drain_timeout: float) → None[source]#: Receive and execute deliveries until cancelled.

RabbitMQ#

class litestar_queues.execution.rabbitmq.config.RabbitMQExecutionConfig(amqp_url: str, queue_name: str | None = None, declare_queue: bool = True, dispatch_stale_after: int = 60, api_timeout: float = 30, delayed_retry_type: Literal['disabled', 'all', 'failed', 'returned'] = 'returned', delayed_retry_min_ms: int = 1000, delayed_retry_max_ms: int = 30000, consumer_timeout_ms: int | None = None)[source]#

Bases: object

RabbitMQ execution-dispatch configuration.

backend_name: ClassVar[str] = 'rabbitmq'#

amqp_url: str#: AMQP or AMQPS connection URL; excluded from representations and telemetry.

queue_name: str | None#: Broker queue name, derived from the queue namespace when omitted.

declare_queue: bool#: Declare the quorum queue when true; otherwise verify it passively.

dispatch_stale_after: int#: Age in seconds after which an owned attempt can be rotated and republished.

api_timeout: float#: Timeout in seconds for connection and publish operations.

delayed_retry_type: Literal['disabled', 'all', 'failed', 'returned']#: RabbitMQ 4.3 quorum-queue delayed retry category.

__init__(amqp_url: str, queue_name: str | None = None, declare_queue: bool = True, dispatch_stale_after: int = 60, api_timeout: float = 30, delayed_retry_type: Literal['disabled', 'all', 'failed', 'returned'] = 'returned', delayed_retry_min_ms: int = 1000, delayed_retry_max_ms: int = 30000, consumer_timeout_ms: int | None = None) → None#

delayed_retry_min_ms: int#: Minimum broker-managed redelivery delay in milliseconds.

delayed_retry_max_ms: int#: Maximum broker-managed redelivery delay in milliseconds.

consumer_timeout_ms: int | None#: Optional acknowledgement timeout; must exceed legitimate task duration.

class litestar_queues.execution.rabbitmq.backend.RabbitMQExecutionBackend(config: QueueConfig | None = None, *, execution_config: RabbitMQExecutionConfig | None = None, connection: Any | None = None)[source]#

Bases: BaseConsumerExecutionBackend

Dispatch bare task identifiers through a RabbitMQ quorum queue.

__init__(config: QueueConfig | None = None, *, execution_config: RabbitMQExecutionConfig | None = None, connection: Any | None = None) → None[source]#: Initialize the execution backend.

property is_external: bool#: Whether this backend dispatches records to another process.

property execution_config: RabbitMQExecutionConfig#

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Dispatch a queue record to an external executor.

Returns:: The external execution reference, if one was created.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#

Recreate deliveries this backend owns that its transport no longer holds.

Parameters:

service – The queue service whose records to repair.
limit – Ceiling on how many records one pass may examine. Bounded maintenance is the only caller, and it always passes a positive budget it needs back.

Returns:

An empty result.

async execute(service: QueueService, record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#: Execute a queue record.

async run_consumer(service: QueueService, *, max_concurrency: int, drain_timeout: float) → None[source]#: Receive and execute deliveries until cancelled.

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

Kafka#

class litestar_queues.execution.kafka.config.KafkaExecutionConfig(bootstrap_servers: str, topic: str = 'litestar-queues', consumer_group: str = 'litestar-queues', dispatch_stale_after: int = 60, api_timeout: float = 30, producer_options: dict[str, ~typing.Any] = <factory>, consumer_options: dict[str, ~typing.Any] = <factory>)[source]#

Bases: object

Kafka execution-dispatch configuration.

backend_name: ClassVar[str] = 'kafka'#

bootstrap_servers: str#: Comma-separated Kafka bootstrap servers.

topic: str#: Topic receiving task-id dispatch records.

consumer_group: str#: Consumer group used by litestar queues run-consumer.

dispatch_stale_after: int#: Age after which a reserved attempt can be rotated and republished.

api_timeout: float#: Timeout applied to producer delivery acknowledgements.

__init__(bootstrap_servers: str, topic: str = 'litestar-queues', consumer_group: str = 'litestar-queues', dispatch_stale_after: int = 60, api_timeout: float = 30, producer_options: dict[str, ~typing.Any] = <factory>, consumer_options: dict[str, ~typing.Any] = <factory>) → None#

producer_options: dict[str, Any]#: Additional AIOKafkaProducer options, such as TLS or SASL settings.

consumer_options: dict[str, Any]#: Additional AIOKafkaConsumer options, such as TLS or SASL settings.

class litestar_queues.execution.kafka.backend.KafkaExecutionBackend(config: QueueConfig | None = None, *, execution_config: KafkaExecutionConfig | None = None, producer: KafkaProducer | None = None, consumer: KafkaConsumer | None = None)[source]#

Bases: BaseConsumerExecutionBackend

Dispatch bare task identifiers through a Kafka consumer group.

__init__(config: QueueConfig | None = None, *, execution_config: KafkaExecutionConfig | None = None, producer: KafkaProducer | None = None, consumer: KafkaConsumer | None = None) → None[source]#: Initialize the execution backend.

property is_external: bool#: Whether this backend dispatches records to another process.

property execution_config: KafkaExecutionConfig#

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#

Dispatch a queue record to an external executor.

Returns:: The external execution reference, if one was created.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#

Recreate deliveries this backend owns that its transport no longer holds.

Parameters:

service – The queue service whose records to repair.
limit – Ceiling on how many records one pass may examine. Bounded maintenance is the only caller, and it always passes a positive budget it needs back.

Returns:

An empty result.

async execute(service: QueueService, record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#: Execute a queue record.

async run_consumer(service: QueueService, *, max_concurrency: int, drain_timeout: float) → None[source]#: Receive and execute deliveries until cancelled.

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

Google Cloud Pub/Sub#

class litestar_queues.execution.pubsub.config.PubSubExecutionConfig(project_id: str, topic_id: str, subscription_id: str, ack_deadline: int = 60, ack_extension_interval: int = 30, dispatch_stale_after: int = 60, api_timeout: float = 30, api_endpoint: str | None = None, api_insecure: bool = False)[source]#

Bases: object

Google Cloud Pub/Sub execution-dispatch configuration.

backend_name: ClassVar[str] = 'pubsub'#

project_id: str#: Google Cloud project containing the topic and subscription.

topic_id: str#: Topic receiving task-id dispatch messages.

subscription_id: str#: Pull subscription consumed by litestar queues run-consumer.

ack_deadline: int#: Initial acknowledgment deadline in seconds; never the queue lease.

ack_extension_interval: int#: Courtesy acknowledgment-deadline extension cadence during execution.

dispatch_stale_after: int#: Age after which a reserved attempt can be rotated and republished.

api_timeout: float#: Timeout applied to unary Pub/Sub API calls.

__init__(project_id: str, topic_id: str, subscription_id: str, ack_deadline: int = 60, ack_extension_interval: int = 30, dispatch_stale_after: int = 60, api_timeout: float = 30, api_endpoint: str | None = None, api_insecure: bool = False) → None#

api_endpoint: str | None#: Optional API host override, including the official emulator host.

api_insecure: bool#: Use plaintext gRPC for a configured local emulator endpoint.

property topic_path: str#: Return the fully qualified Pub/Sub topic path.

property subscription_path: str#: Return the fully qualified Pub/Sub subscription path.

class litestar_queues.execution.pubsub.backend.PubSubExecutionBackend(config: QueueConfig | None = None, *, execution_config: PubSubExecutionConfig | None = None, publisher: PubSubPublisherClient | None = None, subscriber: PubSubSubscriberClient | None = None)[source]#

Bases: BaseConsumerExecutionBackend

Dispatch bare task identifiers through Google Cloud Pub/Sub.

__init__(config: QueueConfig | None = None, *, execution_config: PubSubExecutionConfig | None = None, publisher: PubSubPublisherClient | None = None, subscriber: PubSubSubscriberClient | None = None) → None[source]#: Initialize the execution backend.

property is_external: bool#: Return whether execution occurs in another process.

property execution_config: PubSubExecutionConfig#: Return the effective Pub/Sub configuration.

async dispatch(service: QueueService, record: QueuedTaskRecord) → str | None[source]#: Reserve and publish one task-id delivery.

async repair(service: QueueService, *, limit: int) → DispatchRepairResult[source]#: Rotate and republish stale attempt references.

async execute(service: QueueService, record: QueuedTaskRecord, *, worker_id: str | None = None) → QueuedTaskRecord[source]#: Dispatch a record and return its latest persisted state.

async close() → None[source]#: Close clients created by this backend.

async run_consumer(service: QueueService, *, max_concurrency: int, drain_timeout: float) → None[source]#: Consume streaming-pull deliveries until cancelled.

Consumer#

The programmatic entry point for an external executor: run one queued record by id and exit with a deterministic code. This is the in-process twin of litestar queues run-task – use it from a serverless handler or a custom runner that cannot shell out. The live queue record stays authoritative; the consumer re-fetches it by id and fences on the retry count at claim time.

Public, framework-agnostic consumer API for external execution backends.

run_task / consume_one / TaskExitCode are the programmatic twin of litestar queues run-task: run one queued record by id on any external executor (a Cloud Run Job, a serverless handler, a custom runner) and exit with a deterministic code. Click-free on purpose so broker consumers and in-process handlers can import them without pulling click into the module graph (see test_plugin_lifecycle import boundary).

class litestar_queues.consumer.TaskExitCode(value)[source]#

Bases: IntEnum

Deterministic external-consumer process exit codes.

CANCELLED reports a record whose durable execution result is cancelled. A consumer whose own caller was cancelled reports nothing at all: the cancellation propagates instead, because the record is still running and no outcome has been reached.

SUCCESS = 0#

FAILURE = 1#

MISSING_TASK_ID = 2#

INVALID_TASK_ID = 3#

MISSING_RECORD = 4#

UNKNOWN_TASK = 5#

CLAIM_LOST = 6#

CANCELLED = 7#

MISSING_CONFIG_FACTORY = 8#

async litestar_queues.consumer.consume_one(queue: QueueService, task_id: UUID, *, expected_retry_count: int | None = None, expected_execution_ref: str | None = None) → TaskExitCode[source]#

Claim, execute, and report one queued record identified by its id.

The live record in the queue backend is authoritative; the id only locates it. Redelivery is fenced by the live expected_retry_count at claim time.

The claim is taken before the task name is resolved, because retiring a record this process cannot run is itself a write, and a persistent backend only accepts one over a record that is running and owned.

Returns:: A deterministic task exit code.

Resolve a service and run one queued task by id.

The prefix-aware environment is the default source for every input; the override arguments take precedence over it. config_factory replaces the CONFIG_FACTORY env var, task_id replaces the TASK_ID value, and task_modules replaces TASK_MODULES.

Returns:: A deterministic task exit code.