feat(events): enhance event backend configuration and runtime hints for AsyncPG, Psqlpy, and Psycopg adapters

Author: cofin

docs/guides/adapters/asyncpg.md

@@ -248,8 +248,11 @@ For comprehensive examples and migration guides, see:
 - Publishing uses `connection.notify()` under the hood; consumers rely on
   `connection.add_listener()` with dedicated connections so the shared pool
   stays available for transactional work.
+- For durability and retries, set `driver_features["events_backend"] =
+  "listen_notify_durable"` and include the `events` extension migrations.
 - Force the durable queue fallback (for deterministic testing or multi-tenant
-  workloads) by overriding `driver_features["events_backend"] = "table_queue"`.
+  workloads) by overriding `driver_features["events_backend"] = "table_queue"`
+  and including the `events` migrations.
 
 ## Common Issues
 

docs/guides/adapters/psqlpy.md

@@ -254,12 +254,16 @@ For comprehensive examples and migration guides, see:
 
 ## Event Channels
 
-- Psqlpy adapters use the queue-backed event channel today. Include the
-  `events` extension migrations and call `spec.event_channel(config)` to
-  publish/consume with retryable leases.
-- Psqlpy exposes a built-in `Listener` class. When the native backend ships,
-  SQLSpec will wrap that listener automatically; until then, prefer the durable
-  queue backend for portability across adapters.
+- Psqlpy enables native LISTEN/NOTIFY by default
+  (`driver_features["events_backend"] = "listen_notify"`). Call
+  `spec.event_channel(config)` to publish or consume without migrations.
+- Native listeners use the `Listener` API and a dedicated connection so the
+  shared pool remains available for normal queries.
+- For durability and retries, set `driver_features["events_backend"] =
+  "listen_notify_durable"` and include the `events` extension migrations.
+- The queue-only fallback remains available by setting
+  `driver_features["events_backend"] = "table_queue"` alongside the
+  `events` migrations.
 
 ## Best Practices
 

docs/guides/adapters/psycopg.md

@@ -133,16 +133,16 @@ For comprehensive examples and migration guides, see:
 
 ## Event Channels
 
-- Psycopg currently routes `EventChannel` through the durable queue backend by
-  default (`driver_features["events_backend"] = "table_queue"`). Include the
-  `events` extension migrations, then call `spec.event_channel(config)` to
-  publish/consume events.
-- Native LISTEN/NOTIFY support is coming soon; until then you can still
-  subscribe to channels via the durable queue and rely on leases/acks for retry
-  safety.
-- When you migrate to the native backend, set
-  `driver_features["events_backend"] = "listen_notify"` to opt in once the
-  backend lands.
+- Psycopg enables native LISTEN/NOTIFY support by default
+  (`driver_features["events_backend"] = "listen_notify"`). Call
+  `spec.event_channel(config)` to publish or consume without migrations.
+- Listeners run on dedicated connections so the pool remains available for
+  transactional work.
+- For durability and retries, set `driver_features["events_backend"] =
+  "listen_notify_durable"` and include the `events` extension migrations.
+- The queue-only fallback remains available by setting
+  `driver_features["events_backend"] = "table_queue"` alongside the
+  `events` migrations.
 
 ## Common Issues
 

docs/guides/adapters/spanner.md

@@ -8,26 +8,26 @@ This guide provides specific instructions for the `spanner` adapter.
 
 ## Key Information
 
--   **Driver:** `google-cloud-spanner`
--   **Parameter Style:** `named` with `@` prefix (e.g., `@name`)
--   **Dialect:** `spanner` (custom dialect extending BigQuery)
--   **Transactional DDL:** Not supported (DDL uses separate admin operations)
+- **Driver:** `google-cloud-spanner`
+- **Parameter Style:** `named` with `@` prefix (e.g., `@name`)
+- **Dialect:** `spanner` (custom dialect extending BigQuery)
+- **Transactional DDL:** Not supported (DDL uses separate admin operations)
 
 ## Parameter Profile
 
--   **Registry Key:** `"spanner"`
--   **JSON Strategy:** `helper`
--   **Default Style:** `NAMED_AT` (parameters prefixed with `@`)
+- **Registry Key:** `"spanner"`
+- **JSON Strategy:** `helper`
+- **Default Style:** `NAMED_AT` (parameters prefixed with `@`)
 
 ## Features
 
--   **Full ACID Transactions:** Spanner provides global transactions with strong consistency
--   **Interleaved Tables:** Physical co-location of parent-child rows for performance
--   **Row-Level TTL:** Automatic row expiration via TTL policies
--   **Session Pooling:** Built-in session pool management
--   **UUID Handling:** Automatic UUID-to-bytes conversion
--   **BYTES Handling:** Automatic base64 encoding/decoding for BYTES columns
--   **JSON Support:** Native JSON type handling
+- **Full ACID Transactions:** Spanner provides global transactions with strong consistency
+- **Interleaved Tables:** Physical co-location of parent-child rows for performance
+- **Row-Level TTL:** Automatic row expiration via TTL policies
+- **Session Pooling:** Built-in session pool management
+- **UUID Handling:** Automatic UUID-to-bytes conversion
+- **BYTES Handling:** Automatic base64 encoding/decoding for BYTES columns
+- **JSON Support:** Native JSON type handling
 
 ## Configuration
 
@@ -124,9 +124,31 @@ with config.provide_session() as session:
 ```
 
 The encoding/decoding is transparent - you work with Python bytes directly. The adapter uses:
+
 - `bytes_to_spanner()` - Base64-encode bytes for storage
 - `spanner_to_bytes()` - Base64-decode bytes from storage
 
+### JSON Handling
+
+Spanner JSON parameters should use Python structures instead of pre-serialized strings. SQLSpec wraps JSON
+values with the Spanner client `JsonObject` when available.
+
+- Dicts are treated as JSON objects.
+- Nested sequences (lists/tuples containing dicts or nested lists) are treated as JSON arrays.
+- Flat sequences of scalars are preserved as ARRAY parameters.
+
+If you need a JSON array of scalars or a scalar JSON value, wrap it explicitly:
+
+```python
+from sqlspec.adapters.spanner import spanner_json
+
+payload = spanner_json([1, 2, 3])
+session.execute(
+    "INSERT INTO events (payload_json) VALUES (@payload)",
+    {"payload": payload},
+)
+```
+
 ### UUID Handling
 
 UUIDs are automatically converted to 16-byte BYTES(16) format:
@@ -210,6 +232,7 @@ CREATE TABLE orders (
 ```
 
 Interleaved tables provide:
+
 - Automatic co-location of related data
 - Efficient joins between parent and child tables
 - Cascading deletes for data integrity
@@ -317,17 +340,17 @@ events = store.list_events(session.id)
 
 ## Common Issues
 
--   **DDL Operations:** DDL statements (CREATE TABLE, ALTER TABLE, etc.) cannot be executed through the driver's `execute()` method. Use `database.update_ddl()` for DDL operations.
+- **DDL Operations:** DDL statements (CREATE TABLE, ALTER TABLE, etc.) cannot be executed through the driver's `execute()` method. Use `database.update_ddl()` for DDL operations.
 
--   **Mutation Limit:** Spanner has a 20,000 mutation limit per transaction. For bulk inserts, batch operations into multiple transactions.
+- **Mutation Limit:** Spanner has a 20,000 mutation limit per transaction. For bulk inserts, batch operations into multiple transactions.
 
--   **Read-Only Snapshots:** The default session context uses read-only snapshots. For write operations, use `database.run_in_transaction()` or configure a transaction context.
+- **Read-Only Snapshots:** The default session context uses read-only snapshots. For write operations, use `database.run_in_transaction()` or configure a transaction context.
 
--   **Emulator Limitations:** The Spanner emulator doesn't support all features (e.g., some complex queries, backups). Test critical functionality against a real Spanner instance.
+- **Emulator Limitations:** The Spanner emulator doesn't support all features (e.g., some complex queries, backups). Test critical functionality against a real Spanner instance.
 
--   **`google.api_core.exceptions.AlreadyExists`:** Resource already exists. Check if the table or index already exists before creating.
+- **`google.api_core.exceptions.AlreadyExists`:** Resource already exists. Check if the table or index already exists before creating.
 
--   **`google.api_core.exceptions.NotFound`:** Resource not found. Verify the instance, database, and table names are correct.
+- **`google.api_core.exceptions.NotFound`:** Resource not found. Verify the instance, database, and table names are correct.
 
 ## Best Practices
 

docs/guides/events/database-event-channels.md

@@ -213,7 +213,7 @@ against accidental sync usage.
 
 | Adapter | Backend | Default poll interval | Lease window | Locking hints |
 | --- | --- | --- | --- | --- |
-| AsyncPG / Psycopg / Psqlpy | Native LISTEN/NOTIFY | `N/A` (native notifications) | `N/A` | Dedicated listener connections reuse the driver's native APIs. |
+| AsyncPG / Psycopg / Psqlpy | `listen_notify` (default), `listen_notify_durable` | `0.5s` (queue-backed paths only) | `30s` (queue-backed paths only) | Queue backends add `FOR UPDATE SKIP LOCKED`; native backends use dedicated listener connections. |
 | Oracle | `advanced_queue` (sync adapters) | `aq_wait_seconds` (default `5s`) | `N/A` – AQ removes messages when dequeued | Exposes AQ dequeue options via `extension_config`. |
 | Asyncmy (MySQL) | Queue fallback | `0.25s` | `5s` | Adds `FOR UPDATE SKIP LOCKED` to reduce contention. |
 | DuckDB | Queue fallback | `0.15s` | `15s` | Favor short leases/poll windows so embedded engines do not spin. |
@@ -295,6 +295,10 @@ config = AsyncpgConfig(
 )
 ```
 
+Queue-backed backends (`listen_notify_durable`, `table_queue`) require the
+events extension migrations. Ensure `migration_config["include_extensions"]`
+contains `"events"` before publishing or consuming.
+
 ### Event Queue Stores
 
 Stores generate adapter-specific DDL for the queue table. Each adapter has
@@ -310,8 +314,8 @@ Example store implementations:
 | Adapter | Payload Type | Timestamp Type | Special Handling |
 | --- | --- | --- | --- |
 | AsyncPG / Psycopg / Psqlpy | JSONB | TIMESTAMPTZ | Standard PostgreSQL |
-| Oracle | CLOB | TIMESTAMP | PL/SQL exception blocks for idempotent DDL |
-| MySQL / Asyncmy | JSON | DATETIME(6) | FOR UPDATE SKIP LOCKED |
+| Oracle | JSON or BLOB | TIMESTAMP | PL/SQL exception blocks, auto-detected JSON storage |
+| MySQL / Asyncmy | JSON | DATETIME(6) | Conditional index creation via procedural SQL |
 | DuckDB | JSON | TIMESTAMP | Short poll intervals |
 | BigQuery | JSON | TIMESTAMP | CLUSTER BY for partitioning |
 | Spanner | JSON | TIMESTAMP | Separate DDL execution (no IF NOT EXISTS) |

sqlspec/adapters/asyncpg/config.py

@@ -21,6 +21,7 @@
 )
 from sqlspec.config import AsyncDatabaseConfig, ExtensionConfigs
 from sqlspec.exceptions import ImproperConfigurationError, MissingDependencyError
+from sqlspec.extensions.events._hints import EventRuntimeHints
 from sqlspec.typing import ALLOYDB_CONNECTOR_INSTALLED, CLOUD_SQL_CONNECTOR_INSTALLED, PGVECTOR_INSTALLED
 from sqlspec.utils.config_normalization import apply_pool_deprecations, normalize_connection_config
 from sqlspec.utils.serializers import from_json, to_json
@@ -490,3 +491,8 @@ def get_signature_namespace(self) -> "dict[str, Any]":
             "AsyncpgPreparedStatement": AsyncpgPreparedStatement,
         })
         return namespace
+
+    def get_event_runtime_hints(self) -> "EventRuntimeHints":
+        """Return polling defaults for PostgreSQL queue fallback."""
+
+        return EventRuntimeHints(poll_interval=0.5, select_for_update=True, skip_locked=True)

sqlspec/adapters/asyncpg/events/backend.py

@@ -10,7 +10,7 @@
 from sqlspec.core import SQL
 from sqlspec.exceptions import EventChannelError, ImproperConfigurationError
 from sqlspec.extensions.events import EventMessage
-from sqlspec.extensions.events._queue import QueueEventBackend, TableEventQueue
+from sqlspec.extensions.events._queue import QueueEventBackend, TableEventQueue, build_queue_backend
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.serializers import from_json, to_json
 
@@ -363,16 +363,7 @@ def create_event_backend(
         except ImproperConfigurationError:
             return None
     if backend_name == "listen_notify_durable":
-        queue = TableEventQueue(
-            config,
-            queue_table=extension_settings.get("queue_table"),
-            lease_seconds=extension_settings.get("lease_seconds"),
-            retention_seconds=extension_settings.get("retention_seconds"),
-            select_for_update=extension_settings.get("select_for_update"),
-            skip_locked=extension_settings.get("skip_locked"),
-            json_passthrough=extension_settings.get("json_passthrough"),
-        )
-        queue_backend = QueueEventBackend(queue)
+        queue_backend = build_queue_backend(config, extension_settings, adapter_name="asyncpg")
         try:
             return AsyncpgHybridEventsBackend(config, queue_backend)
         except ImproperConfigurationError:

sqlspec/adapters/psqlpy/config.py

@@ -275,4 +275,4 @@ def get_signature_namespace(self) -> "dict[str, Any]":
     def get_event_runtime_hints(self) -> "EventRuntimeHints":
         """Return LISTEN/NOTIFY defaults for Psqlpy adapters."""
 
-        return EventRuntimeHints(json_passthrough=True)
+        return EventRuntimeHints(poll_interval=0.5, select_for_update=True, skip_locked=True, json_passthrough=True)

sqlspec/adapters/psqlpy/events/backend.py

@@ -9,7 +9,7 @@
 from sqlspec.core import SQL
 from sqlspec.exceptions import EventChannelError, ImproperConfigurationError
 from sqlspec.extensions.events import EventMessage
-from sqlspec.extensions.events._queue import QueueEventBackend, TableEventQueue
+from sqlspec.extensions.events._queue import QueueEventBackend, build_queue_backend
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.serializers import from_json, to_json
 
@@ -319,16 +319,7 @@ def create_event_backend(
         except ImproperConfigurationError:
             return None
     if backend_name == "listen_notify_durable":
-        queue = TableEventQueue(
-            config,
-            queue_table=extension_settings.get("queue_table"),
-            lease_seconds=extension_settings.get("lease_seconds"),
-            retention_seconds=extension_settings.get("retention_seconds"),
-            select_for_update=extension_settings.get("select_for_update"),
-            skip_locked=extension_settings.get("skip_locked"),
-            json_passthrough=extension_settings.get("json_passthrough"),
-        )
-        queue_backend = QueueEventBackend(queue)
+        queue_backend = build_queue_backend(config, extension_settings, adapter_name="psqlpy")
         try:
             return PsqlpyHybridEventsBackend(config, queue_backend)
         except ImproperConfigurationError:

sqlspec/adapters/psycopg/config.py

@@ -21,6 +21,7 @@
     psycopg_statement_config,
 )
 from sqlspec.config import AsyncDatabaseConfig, ExtensionConfigs, SyncDatabaseConfig
+from sqlspec.extensions.events._hints import EventRuntimeHints
 from sqlspec.typing import PGVECTOR_INSTALLED
 from sqlspec.utils.config_normalization import apply_pool_deprecations, normalize_connection_config
 from sqlspec.utils.serializers import to_json
@@ -299,6 +300,11 @@ def get_signature_namespace(self) -> "dict[str, Any]":
         })
         return namespace
 
+    def get_event_runtime_hints(self) -> "EventRuntimeHints":
+        """Return polling defaults for PostgreSQL queue fallback."""
+
+        return EventRuntimeHints(poll_interval=0.5, select_for_update=True, skip_locked=True)
+
 
 class PsycopgAsyncConfig(AsyncDatabaseConfig[PsycopgAsyncConnection, AsyncConnectionPool, PsycopgAsyncDriver]):
     """Configuration for Psycopg asynchronous database connections with direct field-based configuration."""
@@ -495,3 +501,8 @@ def get_signature_namespace(self) -> "dict[str, Any]":
             "PsycopgPoolParams": PsycopgPoolParams,
         })
         return namespace
+
+    def get_event_runtime_hints(self) -> "EventRuntimeHints":
+        """Return polling defaults for PostgreSQL queue fallback."""
+
+        return EventRuntimeHints(poll_interval=0.5, select_for_update=True, skip_locked=True)