Events

Configuration

Consolidated queue event configuration.

class litestar_queues.events.config.EventDeliveryConfig(buffer: EventBufferConfig | None = <factory>, sinks: tuple[QueueEventSink, ...] = (), max_payload_bytes: int | None = None, payload_size_estimator: QueueEventSizeEstimator | None = None, strict: bool = False, publish_task_channel: bool = True, publish_queue_channel: bool = True, publish_global_lifecycle: bool = False)

Bases: object

Configuration for live queue event delivery.

buffer: EventBufferConfig | None

Optional producer buffer; None delivers every event immediately.

sinks: tuple[QueueEventSink, ...]

Additional live-delivery sinks, invoked in tuple order.

max_payload_bytes: int | None

Optional maximum encoded Channels payload size in bytes.

payload_size_estimator: QueueEventSizeEstimator | None

Optional encoded-size estimator used for Channels payload chunking.

strict: bool

Whether the first delivery failure should propagate.

publish_task_channel: bool

Whether task-scoped events also target the canonical task channel.

publish_queue_channel: bool

Whether task-scoped events also target the canonical queue channel.

publish_global_lifecycle: bool

Whether lifecycle events also target the global channel.

__init__(buffer: EventBufferConfig | None = <factory>, sinks: tuple[QueueEventSink, ...] = (), max_payload_bytes: int | None = None, payload_size_estimator: QueueEventSizeEstimator | None = None, strict: bool = False, publish_task_channel: bool = True, publish_queue_channel: bool = True, publish_global_lifecycle: bool = False) → None

class litestar_queues.events.config.QueueEventsConfig(channels: ChannelsLike | None = None, delivery: EventDeliveryConfig | None = None, stream: EventStreamConfig | None = None, history: EventHistoryConfig | None = None)

Bases: object

Group queue event delivery, streaming, and history capabilities.

__init__(channels: ChannelsLike | None = None, delivery: EventDeliveryConfig | None = None, stream: EventStreamConfig | None = None, history: EventHistoryConfig | None = None) → None

channels: ChannelsLike | None

Explicit shared Channels target; None permits app discovery only.

delivery: EventDeliveryConfig | None

Live delivery configuration; None disables live publishing.

stream: EventStreamConfig | None

Application stream endpoints; None registers no endpoints.

history: EventHistoryConfig | None

Backend-owned event history; None disables persistence.

Models

Typed realtime event models for queue tasks.

class litestar_queues.events.models.QueueEvent(*, type: str, scope: ~typing.Literal['task', 'queue', 'worker', 'global', 'custom'], id: str = <factory>, scope_key: str | None = None, task_id: str | None = None, task_name: str | None = None, queue: str | None = None, worker_id: str | None = None, execution_backend: str | None = None, execution_profile: str | None = None, attempt: int | None = None, sequence: int | None = None, level: str | None = None, message: str | None = None, progress_current: int | float | None = None, progress_total: int | float | None = None, progress_percent: float | None = None, actor: ~litestar_queues.events.models.QueueEventActor | None = None, entity: ~litestar_queues.events.models.QueueEventEntityRef | None = None, payload: dict[str, ~typing.Any] = <factory>, occurred_at: ~datetime.datetime = <factory>, schema_version: int = 1, event_key: str | None = None)

Bases: Struct

Stable event envelope for queue lifecycle, progress, log, and custom events.

The wire format is camelCase. Null-valued top-level fields are preserved so subscribers can rely on a stable schema for intermediate progress and log events. Payload contents are passed through verbatim.

type: str

scope: Literal['task', 'queue', 'worker', 'global', 'custom']

id: str

scope_key: str | None

task_id: str | None

task_name: str | None

queue: str | None

worker_id: str | None

execution_backend: str | None

execution_profile: str | None

attempt: int | None

sequence: int | None

level: str | None

message: str | None

progress_current: int | float | None

progress_total: int | float | None

progress_percent: float | None

actor: QueueEventActor | None

entity: QueueEventEntityRef | None

payload: dict[str, Any]

occurred_at: datetime

schema_version: int

event_key: str | None

to_dict() → dict[str, Any]

Return the stable camelCase JSON-compatible event envelope.

Null-valued top-level fields are preserved so subscribers can rely on a stable schema for intermediate progress and log events. Payload contents are passed through verbatim.

to_json() → bytes

Return the event envelope as camelCase JSON bytes.

classmethod from_dict(data: dict[str, Any]) → QueueEvent

Build an event from a camelCase mapping.

Returns:

The queue event.

classmethod from_json(data: str | bytes | bytearray) → QueueEvent

Build an event from camelCase JSON text or bytes.

Returns:

The queue event.

Raises:

TypeError – If the decoded JSON value is not an object.

class litestar_queues.events.models.QueueEventActor(*, type: str | None = None, id: str | None = None, name: str | None = None)

Bases: Struct

Actor reference for a queue event.

type: str | None

id: str | None

name: str | None

to_dict() → dict[str, Any]

Return the camelCase wire mapping for this actor.

classmethod from_dict(data: dict[str, Any]) → QueueEventActor

Build an actor reference from a camelCase mapping.

Returns:

The actor reference.

class litestar_queues.events.models.QueueEventEntityRef(*, type: str, id: str, name: str | None = None)

Bases: Struct

Entity reference for a queue event.

type: str

id: str

name: str | None

to_dict() → dict[str, Any]

Return the camelCase wire mapping for this entity reference.

classmethod from_dict(data: dict[str, Any]) → QueueEventEntityRef

Build an entity reference from a camelCase mapping.

Returns:

The entity reference.

Publisher

Queue event publisher.

class litestar_queues.events.publisher.EventBufferConfig(batch_size: int = 20, flush_interval: float = 0.5, max_pending: int = 2000, overflow: Literal['drop_oldest', 'drop_newest', 'block', 'error'] = 'drop_oldest')

Bases: object

Producer-side micro-batch buffer for live event delivery.

batch_size: int

Maximum live events delivered in one batch.

flush_interval: float

Maximum delay before flushing a partial live-event batch in seconds.

max_pending: int

Maximum live events waiting in the producer-side buffer.

overflow: Literal['drop_oldest', 'drop_newest', 'block', 'error']

Action taken when the pending live-event limit is reached.

__init__(batch_size: int = 20, flush_interval: float = 0.5, max_pending: int = 2000, overflow: Literal['drop_oldest', 'drop_newest', 'block', 'error'] = 'drop_oldest') → None

class litestar_queues.events.publisher.QueueEventPublisher(sink: QueueEventSink | None = None, *, event_log: _QueueEventHistoryWriter | None = None, event_log_strict: bool = False, buffer_config: EventBufferConfig | None = None, strict: bool = False, publish_task_channel: bool = True, publish_queue_channel: bool = True, publish_global_lifecycle: bool = False, namespace: QueueNamespace | str | None = None, observability_runtime: QueueObservabilityRuntimeProtocol | None = None, transport: str | None = None)

Bases: object

Publish queue events through a configured sink.

__init__(sink: QueueEventSink | None = None, *, event_log: _QueueEventHistoryWriter | None = None, event_log_strict: bool = False, buffer_config: EventBufferConfig | None = None, strict: bool = False, publish_task_channel: bool = True, publish_queue_channel: bool = True, publish_global_lifecycle: bool = False, namespace: QueueNamespace | str | None = None, observability_runtime: QueueObservabilityRuntimeProtocol | None = None, transport: str | None = None) → None

strict

publish_task_channel

publish_queue_channel

publish_global_lifecycle

property sink: QueueEventSink

Configured event sink.

set_event_log(event_log: _QueueEventHistoryWriter, *, strict: bool = False) → None

Attach backend-owned durable event history to this publisher.

set_observability_runtime(runtime: QueueObservabilityRuntimeProtocol) → None

Attach the service-owned runtime used for live delivery metrics.

async publish(event: QueueEvent, *, channels: Sequence[str] | None = None, immediate: bool = False) → None

Publish an event to canonical and explicitly supplied channels.

async flush_buffer() → None

Flush all buffered live events.

start_buffer() → None

Start the live event buffer flush loop.

async stop_buffer() → None

Stop and drain the live event buffer.

resolve_channels(event: QueueEvent, *, channels: Sequence[str] | None = None) → tuple[str, ...]

Return canonical publish channels for an event plus explicit extras.

Buffering and Chunking

Producer-side live event buffering.

class litestar_queues.events.buffer.LiveEventBuffer(config: EventBufferConfig, *, sink_publish: SinkPublish, record_drop: RecordDrop, runtime_logger: logging.Logger | None = None)

Bases: object

Bounded producer-side buffer for live queue event delivery.

__init__(config: EventBufferConfig, *, sink_publish: SinkPublish, record_drop: RecordDrop, runtime_logger: logging.Logger | None = None) → None

async add(event: QueueEvent, channels: Sequence[str]) → None

Add an event to the buffer, applying configured overflow behavior.

async flush(*, key: str | tuple[str, str, str | None] | None = None) → None

Drain all buffered events, or only events matching key.

start() → None

Start the interval flush loop if it is not already running.

async stop() → None

Stop the interval loop and drain all remaining buffered events.

litestar_queues.events.buffer.event_buffer_key(event: QueueEvent) → EventBufferKey

Return the buffer key used for scoped flushes.

Live queue-event transport sizing helpers.

litestar_queues.events.chunking.QueueEventSizeEstimator

alias of Callable[[QueueEvent], int]

litestar_queues.events.chunking.estimate_event_payload_bytes(event: QueueEvent) → int

Return the direct JSON payload size for one queue event.

litestar_queues.events.chunking.split_event_batch_by_size(event: ~litestar_queues.events.models.QueueEvent, *, max_bytes: int, size_estimator: ~collections.abc.Callable[[QueueEvent], int] = <function estimate_event_payload_bytes>) → tuple[QueueEvent, ...]

Split package-owned batch events into complete QueueEvent payloads.

Returns:

The original event when no split is needed, otherwise complete event payloads that each fit inside the configured limit.

Task Context

Task execution context and helper APIs for queue event publishing.

class litestar_queues.events.context.TaskBeatSink(*args, **kwargs)

Bases: Protocol

Receives last-value-wins beat progress for a running task.

record_beat(task_id: str, detail: str | None) → None

Record the latest beat detail reported by task_id.

__init__(*args, **kwargs)

class litestar_queues.events.context.TaskExecutionContext(task_id: str, task_name: str, queue: str, worker_id: str | None, execution_backend: str, execution_profile: str | None, attempt: int, event_publisher: QueueEventPublisher, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None)

Bases: object

Context bound while a queue task is executing.

task_id: str

task_name: str

queue: str

worker_id: str | None

execution_backend: str

execution_profile: str | None

attempt: int

event_publisher: QueueEventPublisher

scope_key: str | None

actor: QueueEventActor | None

entity: QueueEventEntityRef | None

property is_cancelled: bool

Whether durable cancellation has reached this execution.

async wait_cancelled() → None

Wait until durable cancellation reaches this execution.

raise_if_cancelled() → None

Raise JobCancelledError after cancellation is requested.

mark_cancelled() → None

async progress(*, current: float | None = None, total: float | None = None, percent: float | None = None, message: str | None = None, payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish a task progress event.

async log(message: str, *, level: str = 'info', payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish a task log event.

async event(event_type: str, *, message: str | None = None, payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish a custom task event.

async lifecycle(event_type: str, *, message: str | None = None, payload: dict[str, Any] | None = None) → None

Publish a worker-owned lifecycle event.

beat(detail: str | None = None) → None

Record last-value-wins progress for the next heartbeat tick.

async publish(event_type: str, *, level: str | None = None, message: str | None = None, progress_current: float | None = None, progress_total: float | None = None, progress_percent: float | None = None, payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope: QueueEventScope = 'task', scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → QueueEvent

Build and publish an event for this task context.

Returns:

The published queue event.

__init__(task_id: str, task_name: str, queue: str, worker_id: str | None, execution_backend: str, execution_profile: str | None, attempt: int, event_publisher: QueueEventPublisher, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None) → None

litestar_queues.events.context.beat(detail: str | None = None) → None

Record progress through the currently bound task context, if any.

litestar_queues.events.context.bind_beat_sink(sink: TaskBeatSink) → Iterator[TaskBeatSink]

Bind sink to receive TaskExecutionContext.beat() calls.

Yields:

The bound beat sink.

litestar_queues.events.context.bind_task_context(context: TaskExecutionContext) → Iterator[TaskExecutionContext]

Bind context as the current task execution context.

This is the supported entry point for external runtimes adopting the events subpackage standalone. While bound, require_current_task_context() and the module-level publish helpers resolve to context.

Yields:

The bound task execution context.

litestar_queues.events.context.get_current_task_context() → TaskExecutionContext | None

Return the task execution context for the current async context.

async litestar_queues.events.context.publish_task_event(event_type: str, *, message: str | None = None, payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish a custom event through the currently bound task context.

async litestar_queues.events.context.publish_task_log(message: str, *, level: str = 'info', payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish a log event through the currently bound task context.

async litestar_queues.events.context.publish_task_progress(*, current: float | None = None, total: float | None = None, percent: float | None = None, message: str | None = None, payload: dict[str, Any] | None = None, channels: Sequence[str] | None = None, scope_key: str | None = None, actor: QueueEventActor | None = None, entity: QueueEventEntityRef | None = None, immediate: bool = False) → None

Publish progress through the currently bound task context.

litestar_queues.events.context.require_current_task_context() → TaskExecutionContext

Return the current task context or raise if none is bound.

Raises:

RuntimeError – If no task context is bound.

Channels

Channel naming helpers for queue events.

class litestar_queues.events.channels.QueueChannels

Bases: object

Canonical channel name factories for queue event scopes.

prefix: ClassVar[str] = 'litestar_queues'

classmethod task(task_id: str, *, topic: str = 'events', namespace: QueueNamespace | str | None = None) → str

Return the channel for task-scoped events.

classmethod queue(queue: str, *, topic: str = 'events', namespace: QueueNamespace | str | None = None) → str

Return the channel for queue-scoped events.

classmethod worker(worker_id: str, *, topic: str = 'events', namespace: QueueNamespace | str | None = None) → str

Return the channel for worker-scoped events.

classmethod global_channel(*, topic: str = 'events', namespace: QueueNamespace | str | None = None) → str

Return the global queue event channel.

classmethod custom(scope_key: str, *, topic: str = 'events', namespace: QueueNamespace | str | None = None) → str

Return a custom queue event channel.

Sinks

Queue event sink protocols and core implementations.

class litestar_queues.events.sinks.CompositeQueueEventSink(sinks: Sequence[QueueEventSink], *, strict: bool = False, namespace: QueueNamespace | str | None = None)

Bases: object

Deliver events to multiple sinks in deterministic order.

__init__(sinks: Sequence[QueueEventSink], *, strict: bool = False, namespace: QueueNamespace | str | None = None) → None

property sinks: tuple[QueueEventSink, ...]

Configured sinks in delivery order.

async open() → None

Open child sinks in declaration order.

async close() → None

Close opened child sinks in reverse declaration order.

async publish(event: QueueEvent, *, channels: Sequence[str]) → None

Publish to every sink, continuing after non-strict failures.

async publish_many(batch: Sequence[tuple[QueueEvent, Sequence[str]]]) → None

Publish a batch to every sink in order.

class litestar_queues.events.sinks.InMemoryQueueEventSink

Bases: object

In-process event sink for tests, examples, and local demos.

__init__() → None

property events: list[QueueEvent]

Published events in publish order.

property published: list[tuple[QueueEvent, tuple[str, ...]]]

Published events with their channels.

events_for(channel: str) → list[QueueEvent]

Return events published to a channel.

async publish(event: QueueEvent, *, channels: Sequence[str]) → None

Store an event in process.

async publish_many(batch: Sequence[tuple[QueueEvent, Sequence[str]]]) → None

Store a batch of events in process.

class litestar_queues.events.sinks.NoopQueueEventSink

Bases: object

Event sink that accepts events and drops them.

async publish(event: QueueEvent, *, channels: Sequence[str]) → None

Drop an event publish.

async publish_many(batch: Sequence[tuple[QueueEvent, Sequence[str]]]) → None

Drop a batch publish.

class litestar_queues.events.sinks.QueueEventSink(*args, **kwargs)

Bases: Protocol

Transport boundary for queue event delivery.

async publish(event: QueueEvent, *, channels: Sequence[str]) → None

Publish an event to the requested channels.

__init__(*args, **kwargs)

async litestar_queues.events.sinks.default_publish_many(sink: QueueEventSink, batch: Sequence[tuple[QueueEvent, Sequence[str]]]) → None

Publish a batch by looping over a sink’s single-event publish method.

Event History

Backend-owned queue event history contracts.

litestar_queues.events.history.RESERVED_EVENT_HISTORY_COLUMNS = frozenset({'actor', 'entity', 'scope', 'scope_key'})

Names held for built-in event-history scoping dimensions.

These are not columns on the table yet. They are reserved so an adopter-declared extra column cannot claim a name the package intends to own.

class litestar_queues.events.history.EventHistoryConfig(batch_size: int = 20, flush_interval: float = 1.0, strict: bool = False, memory_capacity: int = 1000, extra_columns: tuple[~litestar_queues.events.history.EventHistoryExtraColumn, ...] = <factory>)

Bases: object

Configuration for backend-managed queue event history.

batch_size: int

Maximum history records written in one batch.

flush_interval: float

Maximum delay between history batch writes in seconds.

strict: bool

Whether event-history write failures propagate to the publisher.

memory_capacity: int

Maximum retained records for the memory backend.

extra_columns: tuple[EventHistoryExtraColumn, ...]

Adopter-declared scoping columns on the event-history table.

__init__(batch_size: int = 20, flush_interval: float = 1.0, strict: bool = False, memory_capacity: int = 1000, extra_columns: tuple[~litestar_queues.events.history.EventHistoryExtraColumn, ...] = <factory>) → None

class litestar_queues.events.history.EventHistoryExtraColumn(name: str, source: str, indexed: bool = False)

Bases: object

Adopter-declared scoping column on the SQLSpec event-history table.

name: str

Physical column name; must be a valid unquoted SQL identifier.

source: str

Key looked up in the event payload (QueueEvent.payload).

indexed: bool

Whether a (name, occurred_at) index is created.

__init__(name: str, source: str, indexed: bool = False) → None

class litestar_queues.events.history.QueueEventLog(*args, **kwargs)

Bases: Protocol

Backend-owned queue event history writer and query interface.

__init__(*args, **kwargs)

async publish_event(event: QueueEvent) → None

async flush_events() → None

async query_events(query: QueueEventQuery | None = None, *, extra: Mapping[str, str] | None = None) → OffsetPagination[QueueEventLogRecord]

async summarize_stages(query: QueueEventQuery | None = None) → list[QueueEventStageSummary]

async cleanup_events(*, before: datetime, match: QueueEventQuery | None = None, exclude: Sequence[QueueEventQuery] = (), limit: int | None = None) → int

class litestar_queues.events.history.QueueEventLogRecord(event_id: str, event_type: str, task_id: str | None, task_name: str | None, queue: str | None, worker_id: str | None, execution_backend: str | None, execution_profile: str | None, actor_type: str | None, actor_id: str | None, stage: str | None, level: str | None, message: str | None, detail: dict[str, Any], progress_current: float | None, progress_total: float | None, progress_percent: float | None, duration_ms: float | None, sequence: int | None, occurred_at: datetime, created_at: datetime, scope: str | None = None, scope_key: str | None = None, actor: str | None = None, entity: str | None = None, extra: dict[str, str] = <factory>)

Bases: object

A durable queue event history record.

event_id: str

event_type: str

task_id: str | None

task_name: str | None

queue: str | None

worker_id: str | None

execution_backend: str | None

execution_profile: str | None

actor_type: str | None

actor_id: str | None

stage: str | None

level: str | None

message: str | None

detail: dict[str, Any]

progress_current: float | None

progress_total: float | None

progress_percent: float | None

duration_ms: float | None

sequence: int | None

occurred_at: datetime

created_at: datetime

scope: str | None

Envelope scope of the event (task/queue/worker/…).

scope_key: str | None

Adopter scoping key carried on the envelope (tenant, project, account).

actor: str | None

The logical actor.

entity: str | None

Canonical entity key from event_entity_key().

extra: dict[str, str]

__init__(event_id: str, event_type: str, task_id: str | None, task_name: str | None, queue: str | None, worker_id: str | None, execution_backend: str | None, execution_profile: str | None, actor_type: str | None, actor_id: str | None, stage: str | None, level: str | None, message: str | None, detail: dict[str, Any], progress_current: float | None, progress_total: float | None, progress_percent: float | None, duration_ms: float | None, sequence: int | None, occurred_at: datetime, created_at: datetime, scope: str | None = None, scope_key: str | None = None, actor: str | None = None, entity: str | None = None, extra: dict[str, str] = <factory>) → None

class litestar_queues.events.history.QueueEventStageSummary(stage: str | None, event_count: int, total_duration_ms: float, first_event_at: datetime | None, last_event_at: datetime | None, latest_sequence: int | None = None, latest_message: str | None = None, worst_level: str | None = None)

Bases: object

Aggregated queue event history data for a single stage.

stage: str | None

__init__(stage: str | None, event_count: int, total_duration_ms: float, first_event_at: datetime | None, last_event_at: datetime | None, latest_sequence: int | None = None, latest_message: str | None = None, worst_level: str | None = None) → None

event_count: int

total_duration_ms: float

first_event_at: datetime | None

last_event_at: datetime | None

latest_sequence: int | None

Sequence of the newest record in the stage by the stable order key.

latest_message: str | None

Message of the newest record in the stage by the stable order key.

worst_level: str | None

Highest-ranked level present in the stage; None when no record has one.

litestar_queues.events.history.extract_event_extras(payload: Mapping[str, Any] | None, declared_columns: Sequence[EventHistoryExtraColumn]) → dict[str, str]

Extract declared extra columns from an event payload dict.

Returns:

Mapping of physical column names to extracted string values.

litestar_queues.events.history.validate_event_extra_filter(filter_map: Mapping[str, str] | None, declared_columns: Sequence[EventHistoryExtraColumn]) → dict[str, str]

Validate and resolve extra column filter key-value pairs against declared columns.

Returns:

Mapping of resolved physical column names to expected filter values.

Raises:

QueueConfigurationError – If any filter key is not declared in declared_columns.

litestar_queues.events.history.validate_event_history_extra_columns(columns: Sequence[EventHistoryExtraColumn]) → tuple[EventHistoryExtraColumn, ...]

Validate adopter-declared extra event-history columns.

Returns:

The validated declarations as a tuple.

Raises:

QueueConfigurationError – If a name is not a valid unquoted SQL identifier, collides with a package-owned column, uses a reserved scoping-dimension name, repeats another declaration, or the payload source key is empty.

External Producers

Producer facade for queue event publishing.

class litestar_queues.events.producer.QueueEventProducer(publisher: QueueEventPublisher)

Bases: object

Thin facade over a queue event publisher.

__init__(publisher: QueueEventPublisher) → None

task(task_id: str) → _TaskEventHandle

Return a task-scoped event handle.

queue(name: str) → _ScopeEventHandle

Return a queue-scoped event handle.

worker(worker_id: str) → _ScopeEventHandle

Return a worker-scoped event handle.

channel(scope_key: str) → _ScopeEventHandle

Return a custom-channel event handle.

litestar_queues.events.producer.create_event_producer(config: QueueConfig) → _ExternalProducer

Return an external producer context manager for queue event publishing.

Litestar Channels

Litestar Channels event sink for queue events.

class litestar_queues.events.channels_sink.ChannelsQueueEventSink(channels_backend: ChannelsLike, *, manage_lifecycle: bool = False, max_payload_bytes: int | None = None, payload_size_estimator: QueueEventSizeEstimator | None = None)

Bases: object

Event sink that publishes to an app-owned Litestar Channels object.

__init__(channels_backend: ChannelsLike, *, manage_lifecycle: bool = False, max_payload_bytes: int | None = None, payload_size_estimator: QueueEventSizeEstimator | None = None) → None

property channels_backend: ChannelsLike

Wrapped Channels backend or plugin.

property manages_lifecycle: bool

Whether this sink owns its Channels target lifecycle.

async open() → None

Open a worker-owned Channels lifecycle when configured.

async close() → None

Close only the Channels lifecycle opened by this sink.

async publish(event: QueueEvent, *, channels: Sequence[str]) → None

Publish an event to Litestar Channels.

async publish_many(batch: Sequence[tuple[QueueEvent, Sequence[str]]]) → None

Publish grouped events to Litestar Channels.

Stream Configuration and Routes

Configuration for plugin-owned queue event streaming endpoints.

litestar_queues.events.stream_config.ChannelAuthorizer

alias of Callable[[Any, Literal[‘task’, ‘queue’, ‘worker’, ‘global’, ‘custom’], str | None], bool | Awaitable[bool]]

class litestar_queues.events.stream_config.EventStreamConfig(transports: set[~typing.Literal['sse', 'websocket']] = <factory>, path: str = <object object>, guards: list[~collections.abc.Callable[[~typing.Any, ~typing.Any], ~typing.Any]] | None = None, channel_authorizer: ~collections.abc.Callable[[~typing.Any, ~typing.Literal['task', 'queue', 'worker', 'global', 'custom'], str | None], bool | ~collections.abc.Awaitable[bool]] | None = None, unauthenticated_access: ~typing.Literal['warn', 'allow', 'error'] = 'warn', scopes: set[~typing.Literal['task', 'queue', 'worker', 'global', 'custom']] = <factory>, heartbeat_interval: float = 25.0, replay_limit: int = 0, include_in_schema: bool = False, opt: dict[str, ~typing.Any] | None = None)

Bases: object

Configuration for plugin-registered WebSocket queue-event streaming.

transports: set[Literal['sse', 'websocket']]

Enabled browser stream transports.

path: str

Leading-slash route path shared by configured stream transports.

guards: list[Callable[[Any, Any], Any]] | None

Litestar route guards applied to stream endpoints; None adds none.

channel_authorizer: Callable[[Any, Literal['task', 'queue', 'worker', 'global', 'custom'], str | None], bool | Awaitable[bool]] | None

Per-subscription authorizer; None leaves channel selection unrestricted.

unauthenticated_access: Literal['warn', 'allow', 'error']

Policy used when stream endpoints have neither guards nor an authorizer.

__init__(transports: set[~typing.Literal['sse', 'websocket']] = <factory>, path: str = <object object>, guards: list[~collections.abc.Callable[[~typing.Any, ~typing.Any], ~typing.Any]] | None = None, channel_authorizer: ~collections.abc.Callable[[~typing.Any, ~typing.Literal['task', 'queue', 'worker', 'global', 'custom'], str | None], bool | ~collections.abc.Awaitable[bool]] | None = None, unauthenticated_access: ~typing.Literal['warn', 'allow', 'error'] = 'warn', scopes: set[~typing.Literal['task', 'queue', 'worker', 'global', 'custom']] = <factory>, heartbeat_interval: float = 25.0, replay_limit: int = 0, include_in_schema: bool = False, opt: dict[str, ~typing.Any] | None = None) → None

scopes: set[Literal['task', 'queue', 'worker', 'global', 'custom']]

Task-event scopes clients may subscribe to.

heartbeat_interval: float

SSE keepalive interval in seconds; zero disables keepalives.

replay_limit: int

Maximum retained Channels messages replayed on subscription; zero disables replay.

include_in_schema: bool

Whether generated stream routes appear in OpenAPI schema output.

opt: dict[str, Any] | None

Litestar route-handler metadata; None supplies no metadata.

resolve(namespace: QueueNamespace | None = None) → EventStreamConfig

Resolve the namespace-owned default path without mutating this reusable config.

litestar_queues.events.stream_config.Guard

alias of Callable[[Any, Any], Any]

Plugin-owned WebSocket and SSE streaming endpoints for queue events.

Imported only when a stream config is present and enabled, so base package imports stay free of routing and Channels-driver imports.

The router and both relays are private: QueuePlugin is their only caller, and applications configure streaming through EventStreamConfig instead. StreamMetrics stays public because it describes the observability surface an application can record against.

class litestar_queues.events.streaming.StreamMetrics(*args, **kwargs)

Bases: Protocol

Optional metric callbacks used by the WebSocket and SSE stream relays.

on_connect(scope: Literal['task', 'queue', 'worker', 'global', 'custom']) → None

Record a stream connection.

on_event(scope: Literal['task', 'queue', 'worker', 'global', 'custom']) → None

Record an event sent to a stream client.

on_heartbeat(scope: Literal['task', 'queue', 'worker', 'global', 'custom']) → None

Record a heartbeat sent to a stream client.

on_dedup_drop(scope: Literal['task', 'queue', 'worker', 'global', 'custom']) → None

Record a deduplicated event dropped by the stream relay.

on_denial(scope: Literal['task', 'queue', 'worker', 'global', 'custom'], reason: str) → None

Record an authorization denial.

on_disconnect(scope: Literal['task', 'queue', 'worker', 'global', 'custom'], duration_seconds: float) → None

Record stream connection lifetime.

__init__(*args, **kwargs)