Events, Streams & History¶

API reference for event delivery, SSE/WebSocket streaming configuration, durable event history, task execution contexts, standalone producers, and event sinks.

Event & Stream Configuration¶

QueueEventsConfig ¶

Group queue event delivery, streaming, and history capabilities.

Attributes:

Name	Type	Description
`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.

EventDeliveryConfig ¶

Configuration for live queue event delivery.

Attributes:

Name	Type	Description
`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.

EventBufferConfig ¶

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

Attributes:

Name	Type	Description
`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.

EventStreamConfig ¶

Configuration for plugin-registered WebSocket queue-event streaming.

Attributes:

Name	Type	Description
`transports`	`set[EventStreamTransport]`	Enabled browser stream transports.
`path`	`str`	Leading-slash route path shared by configured stream transports.
`guards`	`list[Guard] \| None`	Litestar route guards applied to stream endpoints; `None` adds none.
`channel_authorizer`	`ChannelAuthorizer \| None`	Per-subscription authorizer; `None` leaves channel selection unrestricted.
`unauthenticated_access`	`UnauthenticatedAccess`	Policy used when stream endpoints have neither guards nor an authorizer.
`scopes`	`set[QueueEventScope]`	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.
`max_stream_duration`	`float \| None`	Maximum connection duration in seconds before cleanly closing the stream; `None` disables rollover.
`retry_ms`	`int \| None`	SSE client reconnection delay hint in milliseconds; `None` omits the `retry:` field.
`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 ¶

resolve(namespace=None)

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

Durable Event History & Queries¶

EventHistoryConfig ¶

Configuration for backend-managed queue event history.

Attributes:

Name	Type	Description
`batch_size`	`int`	Maximum history records written in one batch.
`flush_interval`	`float`	Delay before scheduling the oldest pending history batch, 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.
`max_pending`	`int`	Maximum accepted records awaiting persistence or live release.

EventHistoryExtraColumn ¶

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

Attributes:

Name	Type	Description
`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.

QueueEventLog ¶

Bases: Protocol

Backend-owned history with explicit commit acknowledgement and closure.

Custom providers must implement commit-aware publication and closure; accepting an event through publish_event alone is not a commit signal.

publish_event ¶

publish_event(event)

Persist or enqueue a single event into the durable event history.

publish_event_after_commit ¶

publish_event_after_commit(event, *, release, barrier=False)

Release live delivery after commitment, waiting through a barrier when requested.

flush_events ¶

flush_events()

Flush any buffered event history writes to durable storage.

aclose ¶

aclose()

Finish owned history work and release its lifecycle resources.

query_events ¶

query_events(query=None, *, extra=None)

Query persisted event history records with filtering, ordering, and pagination.

summarize_stages ¶

summarize_stages(query=None)

Aggregate matching event history records into per-stage summaries.

cleanup_events ¶

cleanup_events(*, before, match=None, exclude=(), limit=None)

Delete persisted event history records older than before matching the retention filter.

QueueEventLogRecord ¶

A durable queue event history record.

Attributes:

Name	Type	Description
`event_id`	`str`	Unique identifier of the persisted event.
`event_type`	`str`	Event type string (for example `"task.started"` or `"task.progress"`).
`task_id`	`str \| None`	Associated task record identifier, if task-scoped.
`task_name`	`str \| None`	Associated registered task name, if task-scoped.
`queue`	`str \| None`	Logical queue name associated with the event.
`worker_id`	`str \| None`	Worker identifier that emitted the event.
`execution_backend`	`str \| None`	Execution transport name active when the event was emitted.
`execution_profile`	`str \| None`	Optional named execution profile active when the event was emitted.
`actor_type`	`str \| None`	Logical actor category (`"user"`, `"system"`, `"service"`).
`actor_id`	`str \| None`	Identifier of the actor within `actor_type`.
`stage`	`str \| None`	Optional logical pipeline stage reported in the event payload.
`level`	`str \| None`	Severity level (`"debug"`, `"info"`, `"warning"`, `"error"`, `"critical"`).
`message`	`str \| None`	Human-readable summary message carried on the event.
`detail`	`dict[str, Any]`	Structured JSON payload dictionary persisted with the event.
`progress_current`	`float \| None`	Current numeric progress counter reported by the event.
`progress_total`	`float \| None`	Total numeric progress target reported by the event.
`progress_percent`	`float \| None`	Computed or explicit completion percentage (`0.0`-`100.0`).
`duration_ms`	`float \| None`	Optional elapsed duration in milliseconds reported in the event payload.
`sequence`	`int \| None`	Monotonic per-attempt sequence number assigned by the execution context.
`occurred_at`	`datetime`	UTC timestamp when the event was emitted.
`created_at`	`datetime`	UTC timestamp when the history record was persisted.
`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]`	Values extracted for configured `EventHistoryConfig.extra_columns`.

QueueEventStageSummary ¶

Aggregated queue event history data for a single stage.

Attributes:

Name	Type	Description
`stage`	`str \| None`	Stage identifier being summarized, or `None` for unstaged events.
`event_count`	`int`	Total number of persisted events matching this stage.
`total_duration_ms`	`float`	Sum of `duration_ms` across all events in this stage.
`first_event_at`	`datetime \| None`	Earliest `occurred_at` timestamp observed in this stage.
`last_event_at`	`datetime \| None`	Latest `occurred_at` timestamp observed in this stage.
`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.

QueueEventRetentionRule ¶

One ordered, filtered retention rule for durable event history.

Attributes:

Name	Type	Description
`max_age`	`float`	Retention age in seconds; must be finite and greater than 0.
`match`	`QueueEventQuery`	Filter selecting the records this rule governs; an empty query matches all.

QueueEventQuery ¶

Immutable filter, order, and pagination request for durable event history.

Every filter is equality and every set filter is ANDed. None means the dimension is unconstrained.

Attributes:

Name	Type	Description
`task_id`	`str \| None`	Optional task identifier equality filter.
`task_name`	`str \| None`	Optional registered task name equality filter.
`event_type`	`str \| None`	Optional event type equality filter.
`scope`	`str \| None`	Optional event scope equality filter.
`scope_key`	`str \| None`	Optional adopter scope key equality filter.
`entity`	`str \| None`	Optional canonical entity key equality filter.
`level`	`str \| None`	Optional event severity level equality filter.
`order`	`QueueEventOrder`	Sort direction by the stable event history order key (`"asc"` or `"desc"`).
`limit`	`int \| None`	Optional maximum number of records returned per page.
`offset`	`int`	Zero-based record offset applied before `limit`.
`is_paginated`	`bool`	Whether the query constrains ordering output size or position.

filters ¶

filters()

Return the set equality filters as (record attribute, value) pairs.

Returns:

Type	Description
`tuple[tuple[str, str], ...]`	The set filters, in a stable declaration order.

Task Context & Publish Helpers¶

TaskExecutionContext ¶

Context bound while a queue task is executing.

Attributes:

Name	Type	Description
`task_id`	`str`	Hex or UUID string identifier of the executing task record.
`task_name`	`str`	Registered task name being executed.
`queue`	`str`	Logical queue name on which the task is running.
`worker_id`	`str \| None`	Identifier of the worker executing the task, or `None` for inline runs.
`execution_backend`	`str`	Name of the execution backend executing the task.
`execution_profile`	`str \| None`	Optional named execution profile used for this attempt.
`attempt`	`int`	One-based execution attempt number.
`event_publisher`	`QueueEventPublisher`	Event publisher bound to this task execution attempt.
`scope_key`	`str \| None`	Optional default adopter scoping key (tenant, project, account) attached to events.
`actor`	`QueueEventActor \| None`	Optional default actor metadata attached to events published in this context.
`entity`	`QueueEventEntityRef \| None`	Optional default entity reference attached to events published in this context.
`is_cancelled`	`bool`	Whether durable cancellation has reached this execution.

wait_cancelled ¶

wait_cancelled()

Wait until durable cancellation reaches this execution.

raise_if_cancelled ¶

raise_if_cancelled()

Raise JobCancelledError after cancellation is requested.

mark_cancelled ¶

mark_cancelled()

Signal that durable cancellation has been requested for this task attempt.

progress ¶

progress(
    *,
    current=None,
    total=None,
    percent=None,
    message=None,
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
)

Publish a task progress event.

log ¶

log(
    message,
    *,
    level="info",
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
)

Publish a task log event.

event ¶

event(
    event_type,
    *,
    message=None,
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
)

Publish a custom task event.

lifecycle ¶

lifecycle(event_type, *, message=None, payload=None)

Publish a worker-owned lifecycle event.

beat ¶

beat(detail=None)

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

publish ¶

publish(
    event_type,
    *,
    level=None,
    message=None,
    progress_current=None,
    progress_total=None,
    progress_percent=None,
    payload=None,
    channels=None,
    scope="task",
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
)

Build and publish an event for this task context.

Returns:

Type	Description
`QueueEvent`	The published queue event.

TaskBeatSink ¶

Bases: Protocol

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

record_beat ¶

record_beat(task_id, detail)

Record the latest beat detail reported by task_id.

bind_task_context ¶

bind_task_context(context)

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:

Type	Description
`TaskExecutionContext`	The bound task execution context.

bind_beat_sink ¶

bind_beat_sink(sink)

Bind sink to receive TaskExecutionContext.beat() calls.

Yields:

Type	Description
`TaskBeatSink`	The bound beat sink.

get_current_task_context ¶

get_current_task_context()

Return the task execution context for the current async context.

require_current_task_context ¶

require_current_task_context()

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

Raises:

Type	Description
`RuntimeError`	If no task context is bound.

beat ¶

beat(detail=None)

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

publish_task_event ¶

publish_task_event(
    event_type,
    *,
    message=None,
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
    strict=False,
)

Publish a custom event through the currently bound task context, if any.

publish_task_progress ¶

publish_task_progress(
    *,
    current=None,
    total=None,
    percent=None,
    message=None,
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
    strict=False,
)

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

publish_task_log ¶

publish_task_log(
    message,
    *,
    level="info",
    payload=None,
    channels=None,
    scope_key=None,
    actor=None,
    entity=None,
    immediate=False,
    strict=False,
)

Publish a log event through the currently bound task context, if any.

Event Models, Publisher, Producer & Sinks¶

QueueEvent ¶

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.

Attributes:

Name	Type	Description
`type`	`str`	Event type identifier (built-in `QueueEventType` or custom event name).
`scope`	`QueueEventScope`	Routing scope of the event (`"task"`, `"queue"`, `"worker"`, `"global"`, or `"custom"`).
`id`	`str`	Unique hex UUID identifier for this event envelope.
`scope_key`	`str \| None`	Optional tenant, workspace, or custom routing scope key.
`task_id`	`str \| None`	Identifier of the associated task record, when task-scoped.
`task_name`	`str \| None`	Registered name of the associated task, when task-scoped.
`queue`	`str \| None`	Logical queue name associated with the event.
`worker_id`	`str \| None`	Identifier of the worker emitting the event.
`execution_backend`	`str \| None`	Name of the execution backend running the task.
`execution_profile`	`str \| None`	Optional execution profile name active for the task.
`attempt`	`int \| None`	One-based execution attempt counter.
`sequence`	`int \| None`	Monotonic per-attempt event sequence number.
`level`	`str \| None`	Optional log or event severity level.
`message`	`str \| None`	Optional human-readable message accompanying the event.
`progress_current`	`int \| float \| None`	Optional current numeric progress value.
`progress_total`	`int \| float \| None`	Optional total numeric progress target.
`progress_percent`	`float \| None`	Optional completion percentage (`0.0`-`100.0`).
`actor`	`QueueEventActor \| None`	Optional actor attribution reference.
`entity`	`QueueEventEntityRef \| None`	Optional domain entity attribution reference.
`payload`	`dict[str, Any]`	Arbitrary JSON-serializable event payload dictionary.
`occurred_at`	`datetime`	UTC timestamp when the event was created.
`schema_version`	`int`	Wire envelope schema version (`1`).
`event_key`	`str \| None`	Optional idempotency or deduplication key for the event.

to_dict ¶

to_dict()

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 ¶

to_json()

Return the event envelope as camelCase JSON bytes.

from_dict ¶

from_dict(data)

Build an event from a camelCase mapping.

Returns:

Type	Description
`QueueEvent`	The queue event.

from_json ¶

from_json(data)

Build an event from camelCase JSON text or bytes.

Returns:

Type	Description
`QueueEvent`	The queue event.

Raises:

Type	Description
`TypeError`	If the decoded JSON value is not an object.

QueueEventActor ¶

Bases: Struct

Actor reference for a queue event.

Attributes:

Name	Type	Description
`type`	`str \| None`	Logical actor category (for example `"user"`, `"system"`, or `"service"`).
`id`	`str \| None`	Unique identifier of the actor within its category.
`name`	`str \| None`	Optional human-readable display name for the actor.

to_dict ¶

to_dict()

Return the camelCase wire mapping for this actor.

from_dict ¶

from_dict(data)

Build an actor reference from a camelCase mapping.

Returns:

Type	Description
`QueueEventActor`	The actor reference.

QueueEventEntityRef ¶

Bases: Struct

Entity reference for a queue event.

Attributes:

Name	Type	Description
`type`	`str`	Domain entity type name (for example `"document"` or `"deployment"`).
`id`	`str`	Unique domain entity identifier.
`name`	`str \| None`	Optional human-readable display name for the entity.

to_dict ¶

to_dict()

Return the camelCase wire mapping for this entity reference.

from_dict ¶

from_dict(data)

Build an entity reference from a camelCase mapping.

Returns:

Type	Description
`QueueEventEntityRef`	The entity reference.

QueueEventPublisher ¶

Publish queue events through a configured sink.

Attributes:

Name	Type	Description
`sink`	`QueueEventSink`	Configured event sink.

set_event_log ¶

set_event_log(event_log)

Attach backend-owned durable event history to this publisher.

set_observability_runtime ¶

set_observability_runtime(runtime)

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

publish ¶

publish(event, *, channels=None, immediate=False)

Publish an event to canonical and explicitly supplied channels.

flush_buffer ¶

flush_buffer()

Flush all buffered live events.

start_buffer ¶

start_buffer()

Start the live event buffer flush loop.

stop_buffer ¶

stop_buffer()

Stop and drain the live event buffer.

resolve_channels ¶

resolve_channels(event, *, channels=None)

Return canonical publish channels for an event plus explicit extras.

QueueEventProducer ¶

Thin facade over a queue event publisher.

task ¶

task(task_id)

Return a task-scoped event handle.

queue ¶

queue(name)

Return a queue-scoped event handle.

worker ¶

worker(worker_id)

Return a worker-scoped event handle.

channel ¶

channel(scope_key)

Return a custom-channel event handle.

create_event_producer ¶

create_event_producer(config)

Return an external producer context manager for queue event publishing.

QueueEventSink ¶

Bases: Protocol

Transport boundary for queue event delivery.

publish ¶

publish(event, *, channels)

Publish an event to the requested channels.

InMemoryQueueEventSink ¶

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

Attributes:

Name	Type	Description
`events`	`list[QueueEvent]`	Published events in publish order.
`published`	`list[tuple[QueueEvent, tuple[str, ...]]]`	Published events with their channels.

events_for ¶

events_for(channel)

Return events published to a channel.

publish ¶

publish(event, *, channels)

Store an event in process.

publish_many ¶

publish_many(batch)

Store a batch of events in process.

ChannelsQueueEventSink ¶

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

Attributes:

Name	Type	Description
`channels_backend`	`ChannelsLike`	Wrapped Channels backend or plugin.
`manages_lifecycle`	`bool`	Whether this sink owns its Channels target lifecycle.

open ¶

open()

Open a worker-owned Channels lifecycle when configured.

close ¶

close()

Close only the Channels lifecycle opened by this sink.

publish ¶

publish(event, *, channels)

Publish an event to Litestar Channels.

publish_many ¶

publish_many(batch)

Publish grouped events to Litestar Channels.

StreamMetrics ¶

Bases: Protocol

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

on_connect ¶

on_connect(scope)

Record a stream connection.

on_event ¶

on_event(scope)

Record an event sent to a stream client.

on_heartbeat ¶

on_heartbeat(scope)

Record a heartbeat sent to a stream client.

on_dedup_drop ¶

on_dedup_drop(scope)

Record a deduplicated event dropped by the stream relay.

on_denial ¶

on_denial(scope, reason)

Record an authorization denial.

on_disconnect ¶

on_disconnect(scope, duration_seconds)

Record stream connection lifetime.