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.

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 Oracle Advanced Queuing backend uses 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. Native backends remain fire-and-forget; hybrid (listen_notify_durable) backends acknowledge through the durable table queue.

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 iter_events(channel, *, poll_interval=None)

Yield events as they become available.

Return type:

AsyncIterator[EventMessage]

listen(channel, handler, *, 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

iter_events(channel, *, poll_interval=None)

Yield events as they become available.

Return type:

Iterator[EventMessage]

listen(channel, handler, *, 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

final class sqlspec.extensions.events.AsyncTableEventQueue

Bases: _BaseTableEventQueue

Async table queue implementation.

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.

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 _build_create_table_sql() 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]

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, json_passthrough=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 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.

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.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