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,
            "listener_queue_capacity": 256,
        }
    },
)

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.

listener_queue_capacity bounds pending PostgreSQL notifications separately for each consumer. It is unbounded when omitted. At capacity, the oldest pending notification is discarded so the newest notification can be retained. For notify this intentionally loses the transient notification. For notify_queue only the wake-up marker is lost; the durable row remains in the table and reconciliation recovers it. The setting is accepted by every event store so applications can share configuration across adapters, but it does not change poll_queue polling or Oracle AQ and TxEventQ native dequeue behavior.

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.

Listener hubs report aggregate pending depth as events.listener.queue.depth and increment the cumulative events.listener.queue.dropped metric whenever overflow evicts an item. Shutdown clears pending payloads and records depth zero; a restarted hub keeps the runtime's cumulative dropped count.

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. They dequeue directly and do not add the PostgreSQL listener buffer described above. 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.

Notification payload budget

PostgreSQL rejects a NOTIFY payload of 8,000 bytes or more, so SQLSpec publishes at most MAX_NOTIFY_BYTES (7,999) bytes. That budget covers the complete encoded envelope — the event ID, your payload mapping, your metadata mapping, and the publication timestamp — not just the payload you pass to publish(). Sizes are counted in UTF-8 bytes, so accented characters, emoji, and CJK text each consume more than one byte per character.

Use fits_notify_payload() to check a prospective event and measure_notify_payload() to size chunks. Pass your own event_id when you do not let the backend generate the canonical UUID-hex identifier, so the measurement matches exactly what is published:

from sqlspec.extensions.events import (
    MAX_NOTIFY_BYTES,
    fits_notify_payload,
    measure_notify_payload,
)


def chunk_records(records, event_id, metadata=None):
    """Split records into batches that fit one native notification."""
    chunk = []
    for record in records:
        candidate = [*chunk, record]
        if chunk and not fits_notify_payload({"records": candidate}, metadata, event_id=event_id):
            yield chunk
            chunk = [record]
            continue
        chunk = candidate
    if chunk:
        yield chunk


payload = {"records": [{"id": 1}]}
if not fits_notify_payload(payload, event_id="ingest-42"):
    overflow = measure_notify_payload(payload, event_id="ingest-42") - MAX_NOTIFY_BYTES
    raise ValueError(f"event is {overflow} bytes over the notification budget")

Publishing an oversized event raises EventChannelError before any database call, and the message reports both the measured and the maximum byte count.

Large content does not belong in a notification. Write it to a durable table or object store and notify with a compact reference — a row ID, a batch ID, or an object key — that the consumer resolves after it wakes up. For notify_queue, SQLSpec already does this: batch markers carry only marker_id and batch_size, and the durable queue rows remain the source of truth. A transient notify event is never durable regardless of its size, so a notification that fits the budget is still a best-effort wakeup and not a delivery guarantee.

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.MAX_NOTIFY_BYTES = 7999

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by '+' or '-' and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int('0b100', base=0) 4

sqlspec.extensions.events.measure_notify_payload(payload, metadata=None, *, event_id=None)

Return the encoded UTF-8 byte size of the native notification envelope.

The measurement covers the complete envelope rather than only the payload mapping. Omitting event_id measures the canonical backend UUID-hex shape.

Return type:

int

sqlspec.extensions.events.fits_notify_payload(payload, metadata=None, *, event_id=None)

Return whether the native notification envelope fits the PostgreSQL budget.

Return type:

bool

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

Encode event data as JSON for NOTIFY payload.

Raises:

EventChannelError -- If the encoded envelope exceeds the PostgreSQL notification budget.

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