Events

Pub/sub event channel system with database-backed queue support. Provides both sync and async channels with listener management and native backend integration for databases that support LISTEN/NOTIFY.

Transport selection

Choose a transport by delivery semantics:

• notify — transient native notification with no replay or retry.

• notify_queue — durable competing-consumer queue with a native wakeup hint.

• poll_queue — durable competing-consumer queue discovered by polling.

• aq — Oracle Advanced Queuing, with explicit provisioning and privileges.

• txeventq — Oracle Transactional Event Queues, with explicit provisioning and privileges.

The durable queue is the source of truth for notify_queue; native notifications only prompt consumers to check it. Durable event queues are not browser fan-out transports.

Set extension_config["events"]["backend"] to select the transport. The adapter driver_features["events_backend"] value is used only when the extension setting is absent. Retired transport names fail with an explicit canonical replacement instead of silently changing delivery semantics.

Adapter support

Adapter family	Available transports	Default
PostgreSQL (asyncpg, psycopg, psqlpy)	notify, notify_queue, poll_queue	notify
Oracle	poll_queue, aq, txeventq	poll_queue
Other database adapters	poll_queue	poll_queue

Configure the transport and durable reconciliation cadence independently:

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://...", "max_size": 5},
    extension_config={
        "events": {
            "backend": "notify_queue",
            "event_poll_interval": 1.0,
        }
    },
)

event_poll_interval controls how often durable transports reconcile the queue when no native wakeup arrives. The older poll_interval setting is a compatibility input; event_poll_interval takes precedence when both are provided.

polling is not a SQLSpec backend name. Litestar Queues uses it for the fallback worker mode where no push wakeup transport is available and the worker waits for its configured polling interval.

Native LISTEN/NOTIFY model

Native PG event backends (asyncpg, psycopg async/sync, psqlpy) hold a single persistent LISTEN connection per backend instance. Each backend owns its own listener hub that:

• Acquires the dedicated LISTEN connection lazily on first subscribe.

• Emits LISTEN <channel> exactly once per channel and UNLISTEN on unsubscribe / shutdown.

• Dispatches incoming notifications into per-channel asyncio.Queue instances (or queue.Queue for the sync psycopg variant).

• Serializes subscribe / unsubscribe under a lock so concurrent callers cannot race on driver-level statements that share the connection.

The listener lease is held for the backend lifetime. Publishers use separate, short-lived pooled sessions, so a shared PostgreSQL pool must configure at least two connections: max_size >= 2 for asyncpg/psycopg and max_db_pool_size >= 2 for psqlpy. Native backend construction rejects a configured pool of size one instead of allowing publication to deadlock behind the listener.

The Oracle native backends (aq and txeventq) use an analogous pattern: a per-channel queue-handle cache backed by a single dedicated session per backend instance. dequeue honors min(poll_interval, aq_wait_seconds) as its wait bound so the caller's polling cadence is respected.

ack / nack semantics are unchanged. notify remains fire-and-forget; notify_queue acknowledges through the durable table queue.

Batch publication and recovery

Both AsyncEventChannel and SyncEventChannel provide publish_many(events). Each item is a (channel, payload, metadata) tuple, and the returned event IDs preserve input order. Batch-capable implementations commit each grouped call atomically. Backends without publish_many, including the current Oracle native transports, use an ordered single-event fallback; that fallback is not atomic across the batch.

poll_queue bulk-inserts the independent event rows with one publisher session and transaction. PostgreSQL notify publishes the normal per-event notification envelopes in one publisher transaction, so each notification keeps its existing payload and size limit.

For PostgreSQL notify_queue, SQLSpec bulk-inserts all durable rows and then emits one compact marker per channel in the same transaction. A marker contains only marker_id and batch_size; it is a wakeup hint, not a batch event envelope or source of truth. A consumer uses that marker to drain the queued rows without waiting for another notification. Duplicate markers are ignored, and a missing marker is recovered by durable reconciliation on event_poll_interval.

Oracle native event backends

Oracle provides two native messaging backends in addition to the default poll_queue:

• aq — classic Oracle Advanced Queuing (AQ).

• txeventq — Oracle Transactional Event Queues (TxEventQ).

Both share the same client path and JSON payloads; they differ only in how the underlying queue is provisioned. Select one via events.backend:

from sqlspec.adapters.oracledb import OracleAsyncConfig

config = OracleAsyncConfig(
    connection_config={"dsn": "..."},
    extension_config={"events": {"backend": "txeventq"}},
)

The default remains poll_queue, which works on every Oracle edition without extra privileges; both native backends are opt-in.

Requirements

• Thin mode — both backends run in python-oracledb's default Thin mode; no Instant Client / Thick mode is required.

• JSON payloads require Oracle Database 21c or newer (23ai satisfies this).

• Privileges — the connecting user needs DBMS_AQADM access. Grant aq_administrator_role, aq_user_role and EXECUTE ON dbms_aq.

Provisioning

The backend attaches to an existing queue; it does not create one. Provision the queue with DBMS_AQADM first:

• aq — create_queue_table(queue_payload_type => 'JSON') + create_queue + start_queue.

• txeventq — create_transactional_event_queue(queue_payload_type => 'JSON', multiple_consumers => FALSE) + start_queue.

By default all channels route through a single physical queue (SQLSPEC_EVENTS_QUEUE) with the channel carried in the event envelope. To isolate channels onto per-channel physical queues, template the queue name with {channel} via the aq_queue setting (for example "aq_queue": "SQLSPEC_EVT_{channel}") and provision one queue per channel.

Channels

class sqlspec.extensions.events.AsyncEventChannel

Bases: object

Event channel for asynchronous database configurations.

__init__(config)

async publish(channel, payload, metadata=None)

Publish an event to a channel.

Return type:

str

async publish_many(events)

Publish independent events in one grouped operation when supported.

Backend-native implementations are atomic per grouped call. A backend without publish_many uses an ordered single-event fallback, which is not atomic across the full batch.

Return type:

list[str]

iter_events(channel, *, event_poll_interval=None, poll_interval=None)

Yield events as they become available.

Return type:

AsyncIterator[EventMessage]

listen(channel, handler, *, event_poll_interval=None, poll_interval=None, auto_ack=True)

Start an async task that delivers events to handler.

Return type:

AsyncEventListener

async stop_listener(listener_id)

Stop a running listener.

Return type:

None

async ack(event_id)

Acknowledge an event.

Return type:

None

async nack(event_id)

Return an event to the queue for redelivery.

Return type:

None

async shutdown()

Shutdown the event channel and release backend resources.

Return type:

None

class sqlspec.extensions.events.SyncEventChannel

Bases: object

Event channel for synchronous database configurations.

__init__(config)

publish(channel, payload, metadata=None)

Publish an event to a channel.

Return type:

str

publish_many(events)

Publish independent events in one grouped operation when supported.

Backend-native implementations are atomic per grouped call. A backend without publish_many uses an ordered single-event fallback, which is not atomic across the full batch.

Return type:

list[str]

iter_events(channel, *, event_poll_interval=None, poll_interval=None)

Yield events as they become available.

Return type:

Iterator[EventMessage]

listen(channel, handler, *, event_poll_interval=None, poll_interval=None, auto_ack=True)

Start a background thread that invokes handler for each event.

Return type:

SyncEventListener

stop_listener(listener_id)

Stop a running listener.

Return type:

None

ack(event_id)

Acknowledge an event.

Return type:

None

nack(event_id)

Return an event to the queue for redelivery.

Return type:

None

shutdown()

Shutdown the event channel and release backend resources.

Return type:

None

Listeners

class sqlspec.extensions.events.AsyncEventListener

Bases: object

Represents a running async listener task.

async stop()

Signal the listener to stop and await task completion.

Return type:

None

__init__(id, channel, task, stop_event, poll_interval)

class sqlspec.extensions.events.SyncEventListener

Bases: object

Represents a running sync listener thread.

stop()

Signal the listener to stop and join the thread.

Return type:

None

__init__(id, channel, thread, stop_event, poll_interval)

Event Queue

The durable table queue is available for SQL Server through arrow_odbc when configured with Microsoft ODBC Driver 18. It uses SQL Server DATETIME2(6) timestamps and NVARCHAR payload columns.

Durable queue migrations reconcile missing tables and additive columns from the adapter store's canonical DDL. Set events.manage_schema=False when an external migration system owns the queue schema. Set events.create_schema=False to avoid creating an absent queue table. Column renames, drops, and type changes still require an explicit migration.

Queue table storage options

Put durable queue tuning under extension_config["events"]. SQLSpec validates the mapping against the selected adapter; an unknown key or an option that the backend cannot honor raises ImproperConfigurationError.

• PostgreSQL (asyncpg, psycopg, and psqlpy) accepts fillfactor, autovacuum_vacuum_scale_factor, and autovacuum_analyze_scale_factor. These queue-table settings are opt-in.

• BigQuery accepts partitioning, partition_expiration_days, and require_partition_filter for available_at partitioning. Existing channel and status clustering is preserved.

• SQLite and AioSQLite accept pragma_profile and pragma_overrides. PRAGMAs run once during schema preparation rather than on every queue operation.

• Oracle Database accepts compression, partitioning, in_memory, and table_options. See Extension Table Storage Options for how SQLSpec handles optional database capabilities.

CockroachDB deliberately does not expose its session-table row TTL for durable queues. Queue acknowledgement and retention have different semantics, so session-only TTL keys are rejected rather than translated to destructive queue DDL. Other adapters use their existing queue-table defaults and reject these backend-specific storage keys.

For example, configure a PostgreSQL queue for a write-heavy workload:

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://localhost/app"},
    extension_config={
        "events": {
            "backend": "notify_queue",
            "fillfactor": 70,
            "autovacuum_vacuum_scale_factor": 0.05,
            "autovacuum_analyze_scale_factor": 0.02,
        }
    },
)

final class sqlspec.extensions.events.AsyncTableEventQueue

Bases: _BaseTableEventQueue

Async table queue implementation.

async publish_many(events)

Bulk-insert independent events in one transaction.

Return type:

list[str]

async shutdown()

Shutdown the backend (no-op for table queue).

Return type:

None

final class sqlspec.extensions.events.SyncTableEventQueue

Bases: _BaseTableEventQueue

Sync table queue implementation.

publish_many(events)

Bulk-insert independent events in one transaction.

Return type:

list[str]

shutdown()

Shutdown the backend (no-op for table queue).

Return type:

None

sqlspec.extensions.events.build_queue_backend(config, extension_settings=None, *, adapter_name=None, hints=None)

Build a table queue backend using adapter hints and extension overrides.

Return type:

SyncTableEventQueue | AsyncTableEventQueue

Store

class sqlspec.extensions.events.BaseEventQueueStore

Bases: ABC, Generic[ConfigT]

Base class for adapter-specific event queue DDL generators.

This class provides a hook-based pattern for DDL generation. Adapters only need to override _column_types() and optionally any hook methods for dialect-specific variations:

• _string_type(length): String type syntax (default: VARCHAR(N))

• _integer_type(): Integer type syntax (default: INTEGER)

• _timestamp_default(): Timestamp default expression (default: CURRENT_TIMESTAMP)

• _primary_key_syntax(): Inline PRIMARY KEY clause (default: empty, PK on column)

• _table_clause(): Additional table options (default: empty)

For complex dialects (Oracle PL/SQL, BigQuery CLUSTER BY), adapters may override _table_ddl() directly.

__init__(config)

property table_name: str

Return the configured queue table name.

property settings: dict[str, TypeAliasForwardRef('typing.Any')]

Return extension settings for adapters to inspect.

create_statements()

Return statements required to create the queue table and indexes.

Return type:

list[str]

drop_statements()

Return statements required to drop queue artifacts.

Return type:

list[str]

prepare_schema_sync(driver)

Prepare adapter-specific schema decisions with a synchronous driver.

Return type:

None

async prepare_schema_async(driver)

Prepare adapter-specific schema decisions with an asynchronous driver.

Return type:

None

reconcile_schema_sync(driver)

Apply additive queue-table changes with a synchronous driver.

Return type:

SchemaEnsureResult

async reconcile_schema_async(driver)

Apply additive queue-table changes with an asynchronous driver.

Return type:

SchemaEnsureResult

Models

final class sqlspec.extensions.events.EventMessage

Bases: object

Structured payload delivered to event handlers.

__init__(event_id, channel, payload, metadata, attempts, available_at, lease_expires_at, created_at)

final class sqlspec.extensions.events.EventRuntimeHints

Bases: object

Adapter-specific defaults for event polling and leases.

__init__(poll_interval=1.0, lease_seconds=30, retention_seconds=86400, select_for_update=False, skip_locked=False)

Protocols

class sqlspec.extensions.events.AsyncEventBackendProtocol

Bases: Protocol

Protocol for async event backends.

All async event backends (native or queue-based) must implement these methods.

async publish(channel, payload, metadata=None)

Publish an event to a channel.

Parameters:

• channel (str) -- Target channel name.

• payload (dict[str, typing.Any]) -- Event payload (must be JSON-serializable).

• metadata (dict[str, typing.Any] | None) -- Optional metadata dict.

Return type:

str

Returns:

The event ID.

async publish_many(events)

Publish independent events as one grouped backend operation.

Implementations with native batching must preserve input order in the returned event IDs. Backends without native batching are invoked through the event channel's single-event fallback.

Return type:

list[str]

async dequeue(channel, poll_interval)

Dequeue an event from the channel.

Parameters:

• channel (str) -- Channel name to listen on.

• poll_interval (float) -- Timeout in seconds to wait for a notification.

Return type:

EventMessage | None

Returns:

EventMessage if a notification was received, None otherwise.

async ack(event_id)

Acknowledge an event.

Parameters:

event_id (str) -- ID of the event to acknowledge.

Return type:

None

async nack(event_id)

Return an event to the queue for redelivery.

Parameters:

event_id (str) -- ID of the event to return.

Return type:

None

async shutdown()

Shutdown the backend and release resources.

Return type:

None

__init__(*args, **kwargs)

class sqlspec.extensions.events.SyncEventBackendProtocol

Bases: Protocol

Protocol for sync event backends.

All sync event backends (native or queue-based) must implement these methods.

publish(channel, payload, metadata=None)

Publish an event to a channel.

Parameters:

• channel (str) -- Target channel name.

• payload (dict[str, typing.Any]) -- Event payload (must be JSON-serializable).

• metadata (dict[str, typing.Any] | None) -- Optional metadata dict.

Return type:

str

Returns:

The event ID.

publish_many(events)

Publish independent events as one grouped backend operation.

Implementations with native batching must preserve input order in the returned event IDs. Backends without native batching are invoked through the event channel's single-event fallback.

Return type:

list[str]

dequeue(channel, poll_interval)

Dequeue an event from the channel.

Parameters:

• channel (str) -- Channel name to listen on.

• poll_interval (float) -- Timeout in seconds to wait for a notification.

Return type:

EventMessage | None

Returns:

EventMessage if a notification was received, None otherwise.

ack(event_id)

Acknowledge an event.

Parameters:

event_id (str) -- ID of the event to acknowledge.

Return type:

None

nack(event_id)

Return an event to the queue for redelivery.

Parameters:

event_id (str) -- ID of the event to return.

Return type:

None

shutdown()

Shutdown the backend and release resources.

Return type:

None

__init__(*args, **kwargs)

class sqlspec.extensions.events.AsyncEventHandler

Bases: Protocol

Protocol describing async event handler callables.

async __call__(message)

Process a queued event message asynchronously.

Return type:

Any

__init__(*args, **kwargs)

class sqlspec.extensions.events.SyncEventHandler

Bases: Protocol

Protocol describing sync event handler callables.

__call__(message)

Process a queued event message synchronously.

Return type:

Any

__init__(*args, **kwargs)

Payload Helpers

sqlspec.extensions.events.encode_notify_payload(event_id, payload, metadata)

Encode event data as JSON for NOTIFY payload.

Raises:

EventChannelError -- If the encoded payload exceeds PostgreSQL's 8KB limit.

Return type:

str

sqlspec.extensions.events.decode_notify_payload(channel, payload)

Decode JSON payload from NOTIFY into an EventMessage.

Return type:

EventMessage

sqlspec.extensions.events.parse_event_timestamp(value)

Parse a timestamp value into a timezone-aware datetime.

Handles ISO format strings, datetime objects, and falls back to current UTC time for invalid or missing values.

Return type:

datetime

Utility Functions

sqlspec.extensions.events.load_native_backend(config, backend_name, extension_settings, adapter_name=None)

Load adapter-specific native backend if available.

Return type:

Any | None

sqlspec.extensions.events.resolve_poll_interval(poll_interval, default)

Resolve poll interval with validation.

Return type:

float

sqlspec.extensions.events.resolve_event_poll_interval(event_poll_interval, poll_interval, default)

Resolve the event reconciliation interval with compatibility precedence.

Return type:

float

sqlspec.extensions.events.normalize_event_channel_name(name)

Validate event channel identifiers and return normalized name.

Return type:

str

sqlspec.extensions.events.normalize_queue_table_name(name)

Validate schema-qualified identifiers and return normalized name.

Return type:

str