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.

Adapter-specific configuration

Use sqlspec.config.EventsConfig for shared queue and transport settings. Storage tuning belongs to the adapter's exported Events configuration type, inside the same extension_config["events"] dictionary:

from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.asyncpg.events import AsyncpgEventsConfig

events = AsyncpgEventsConfig(
    backend="notify_queue",
    fillfactor=80,
    listener_queue_capacity=256,
)
config = AsyncpgConfig(extension_config={"events": events})

AsyncpgEventsConfig and PsycopgEventsConfig support PostgreSQL storage tuning and native listener capacity. PsqlpyEventsConfig supports the storage tuning fields; its listener does not consume listener_queue_capacity. AiosqliteEventsConfig and SqliteEventsConfig expose the SQLite PRAGMA profile and overrides. Import each from its adapter's events package.

BigQueryEventsConfig.partitioning is a boolean switch for partitioning by DATE(available_at); partition expiration and required-filter settings also enable partitioning. OracleEventsConfig instead accepts structured OracleEventsPartitionConfig and OracleEventsCompressionConfig values, along with Oracle storage and AQ options. Oracle visibility accepts a native integer or a DEQ_IMMEDIATE/DEQ_ON_COMMIT name. Omitting it preserves the driver default, DEQ_ON_COMMIT.

Stores and channels reject settings the selected adapter cannot consume, including native notification and AQ channels. Ordinary dictionaries remain supported. run_migrations is not an Events setting: automatic additive reconciliation uses manage_schema and create_schema; execute versioned migrations explicitly through the migration commands.

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

Both channel classes expose backend_name, the resolved backend kind (notify, notify_queue, poll_queue, aq, or txeventq), and metrics_snapshot(). The snapshot holds every observability metric recorded for the channel's database configuration, including loader, migration, storage, and other event channels on that configuration. Values are floats, and each key is prefixed with the configuration's bind_key, or with the configuration class name when no bind key is set. Counter names vary by backend, so select the event metrics by name rather than assuming a fixed key:

channel = spec.event_channel(config)
await channel.publish("notifications", {"action": "refresh"})

event_metrics = {
    name: value for name, value in channel.metrics_snapshot().items() if ".events." in name
}
print(channel.backend_name, event_metrics)

class sqlspec.extensions.events.AsyncEventChannel

Bases: object

Event channel for asynchronous database configurations.

__init__(config)

property backend_name: str

Return the resolved event backend kind, such as "notify" or "poll_queue".

metrics_snapshot()

Return the observability metrics recorded for this channel's configuration.

Return type:

dict[str, float]

Returns:

A new dictionary of metric values keyed under the configuration's diagnostics prefix.

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)

property backend_name: str

Return the resolved event backend kind, such as "notify" or "poll_queue".

metrics_snapshot()

Return the observability metrics recorded for this channel's configuration.

Return type:

dict[str, float]

Returns:

A new dictionary of metric values keyed under the configuration's diagnostics prefix.

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 poll_queue table transport is supported across all SQLSpec relational and analytical adapters, including:

• ADBC (AdbcEventQueueStore) with dialect-aware DDL and index guards for PostgreSQL, SQLite, DuckDB, BigQuery, Snowflake, Oracle, and IBM Db2.

• SQL Server through mssql_python, pymssql, and arrow_odbc (using DATETIME2(6) timestamps and NVARCHAR(MAX) JSON payload columns).

• IBM Db2 through db2, arrow_odbc, and adbc (using CLOB(1M) payloads and SYSCAT catalog guards).

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 require an explicit versioned 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.

• BigQuery accepts partitioning, partition_expiration_days, and require_partition_filter for available_at partitioning alongside channel and status clustering.

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

• Google Cloud Spanner accepts shard_count (adds a generated shard_id prefix to the primary key and covering index to prevent split hotspotting) and retention_seconds (generates a native ROW DELETION POLICY (OLDER_THAN(acknowledged_at, INTERVAL N DAY)) on GoogleSQL or TTL INTERVAL 'N days' ON acknowledged_at on spangres instead of running a synchronous DELETE on every ack()). Spanner queues also poll with lock-free read-only snapshots (transaction=False) and execute DML writes inside run_in_transaction for automatic Aborted conflict retries.

CockroachDB does not expose its session-table row TTL for durable queues because queue acknowledgement and retention have distinct semantics. Other adapters use their standard queue-table DDL and reject backend-specific storage keys from unrelated database families.

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

Building your own leased queue

When implementing custom leased-row queue tables or competing-consumer workers, SQLSpec exports the core dialect and claim-verification primitives used by SyncTableEventQueue and AsyncTableEventQueue. These pure helpers render dialect-safe row-limiting, locking clauses, and verify leases against rowcount-blind drivers:

Python 3.9+

from datetime import datetime, timedelta, timezone
from sqlspec.extensions.events import (
    claim_verified,
    lock_clause,
    row_limit_clause,
    select_limit_prefix,
)

dialect = "postgres"
top = select_limit_prefix(dialect, 1)
limit = row_limit_clause(dialect, 1)
lock = lock_clause(select_for_update=True, skip_locked=True)

select_sql = (
    f"SELECT {top}id, payload FROM custom_queue WHERE status = 'pending'"
    f"{limit}{lock}"
)

now = datetime.now(timezone.utc)
lease_expires_at = now + timedelta(seconds=30)
won = claim_verified(fetched_row, lease_expires_at)

Python 3.11+

from datetime import datetime, timedelta, UTC
from sqlspec.extensions.events import (
    claim_verified,
    lock_clause,
    row_limit_clause,
    select_limit_prefix,
)

dialect = "postgres"
top = select_limit_prefix(dialect, 1)
limit = row_limit_clause(dialect, 1)
lock = lock_clause(select_for_update=True, skip_locked=True)

select_sql = (
    f"SELECT {top}id, payload FROM custom_queue WHERE status = 'pending'"
    f"{limit}{lock}"
)

now = datetime.now(UTC)
lease_expires_at = now + timedelta(seconds=30)
won = claim_verified(fetched_row, lease_expires_at)

sqlspec.extensions.events.lock_clause(*, select_for_update, skip_locked, dialect=None)

Render the row locking clause for candidate selection.

Db2 (dialect name db2) locks through the isolation clause WITH RS USE AND KEEP UPDATE LOCKS, which is valid on a read-only cursor and so composes with ORDER BY and FETCH FIRST; the selected row stays update-locked until commit.

Parameters:

• select_for_update (bool) -- Whether to lock selected rows with FOR UPDATE.

• skip_locked (bool) -- Whether to skip already locked rows when locking is enabled.

• dialect (str | None) -- Optional SQL dialect identifier.

Return type:

str

Returns:

Locking clause SQL fragment with leading space, or empty string.

sqlspec.extensions.events.row_limit_clause(dialect, n)

Render trailing row limit clause for candidate selection.

Parameters:

• dialect (str) -- SQL dialect identifier.

• n (int) -- Maximum number of rows to return.

Return type:

str

Returns:

Trailing limit clause SQL fragment with leading space, or empty string.

sqlspec.extensions.events.select_limit_prefix(dialect, n)

Render leading SELECT limit prefix for dialects that limit before columns.

Parameters:

• dialect (str) -- SQL dialect identifier.

• n (int) -- Maximum number of rows to return.

Return type:

str

Returns:

Limit prefix clause with trailing space, or empty string.

sqlspec.extensions.events.claim_verified(row, leased_until)

Confirm claim ownership by matching the stored lease against the claimer's token.

Drivers that cannot report rows affected return zero for a successful claim UPDATE, so a zero rowcount alone cannot distinguish a won claim from a lost race. The persisted lease_expires_at value identifies the winning claimer.

Parameters:

• row (dict[str, Any] | None) -- Row dictionary retrieved from the table queue, or None.

• leased_until (datetime) -- Expiration timestamp expected for the winning claim.

Return type:

bool

Returns:

True if the row contains a lease matching leased_until, False otherwise.

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, cleanup_on_ack=True, use_run_in_transaction=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.get_runtime_hints(adapter, config=None)

Return runtime hints provided by the adapter configuration.

Return type:

EventRuntimeHints

sqlspec.extensions.events.resolve_adapter_name(config)

Resolve adapter name from config module path.

Return type:

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

Configuration

class sqlspec.config.EventsConfig

Bases: TypedDict

Configuration options for the events extension.

Use in extension_config["events"].

migrations_path: NotRequired[str | Path]

Directory containing this extension's migrations, or a '<dotted.module>:<subdir>' specification.

Overrides the default sqlspec.extensions.<name> lookup. Setting this auto-includes the extension in migration_config["include_extensions"].

manage_schema: NotRequired[bool]

True.

Type:

Apply additive target-schema reconciliation. Default

create_schema: NotRequired[bool]

True.

Type:

Create the queue table during managed reconciliation. Default

backend: NotRequired[Literal['notify', 'notify_queue', 'poll_queue', 'aq', 'txeventq']]

Backend implementation. PostgreSQL adapters default to 'notify', others to 'poll_queue'.

• notify: Transient PostgreSQL LISTEN/NOTIFY wakeup; not a durable event ledger

• notify_queue: Durable table queue with PostgreSQL LISTEN/NOTIFY wakeups

• poll_queue: Durable table queue discovered by polling

• aq: Oracle Advanced Queuing

• txeventq: Oracle Transactional Event Queues

queue_table: NotRequired[str]

Name of the fallback queue table. Defaults to 'sqlspec_event_queue'.

lease_seconds: NotRequired[int]

Lease duration for claimed events before they can be retried. Defaults to 30 seconds.

retention_seconds: NotRequired[int]

Retention window for acknowledged events before cleanup. Defaults to 86400 (24 hours).

poll_interval: NotRequired[float]

Compatibility alias for event_poll_interval. Defaults to 1.0.

event_poll_interval: NotRequired[float]

Durable event reconciliation interval in seconds. Takes precedence over poll_interval.

select_for_update: NotRequired[bool]

Use SELECT FOR UPDATE locking when claiming events. Defaults to False.

skip_locked: NotRequired[bool]

Use SKIP LOCKED for non-blocking event claims. Defaults to False.

shard_count: NotRequired[int]

Number of hash shards for queue table and index prefixing (e.g. Cloud Spanner). Defaults to 1.