feat(events): standardize sync method naming and enhance event backends

Rename all sync event methods to use _sync suffix for consistency
with async _async suffix pattern across the events extension:
- publish -> publish_sync
- dequeue -> dequeue_sync
- ack -> ack_sync
- iter_events -> iter_events_sync
- shutdown -> shutdown_sync

Also includes fixes:
- Fix unused import in asyncpg events store
- Fix pyright type errors in channel.py extension_settings handling
- Comprehensive test coverage for events extension

Author: cofin

AGENTS.md

@@ -321,6 +321,7 @@ For detailed implementation patterns, consult these guides:
 | Parameter Profiles | `docs/guides/adapters/parameter-profile-registry.md` |
 | Adapter Guides | `docs/guides/adapters/{adapter}.md` |
 | Framework Extensions | `docs/guides/extensions/{framework}.md` |
+| Database Events | `docs/guides/events/database-event-channels.md` |
 | Quick Reference | `docs/guides/quick-reference/quick-reference.md` |
 | Code Standards | `docs/guides/development/code-standards.md` |
 | Implementation Patterns | `docs/guides/development/implementation-patterns.md` |
@@ -857,6 +858,70 @@ if async_configs:
 
 **Reference implementation:** `sqlspec/cli.py` (lines 218-255, 311-724)
 
+### Events Extension Pattern
+
+The events extension provides database-agnostic pub/sub via two layers:
+
+**Backends** handle communication (LISTEN/NOTIFY, Oracle AQ, table polling):
+
+```python
+# Backend selection via driver_features
+class AdapterDriverFeatures(TypedDict):
+    enable_events: NotRequired[bool]
+    events_backend: NotRequired[Literal["listen_notify", "listen_notify_durable", "table_queue", "advanced_queue"]]
+
+# Auto-detection in config __init__
+if "enable_events" not in driver_features:
+    driver_features["enable_events"] = extension_config.get("events") is not None
+if "events_backend" not in driver_features:
+    driver_features["events_backend"] = "listen_notify"  # or adapter-specific default
+```
+
+**Stores** generate adapter-specific DDL for the queue table:
+
+```python
+# In sqlspec/adapters/{adapter}/events/store.py
+class AdapterEventQueueStore(BaseEventQueueStore[AdapterConfig]):
+    __slots__ = ()
+
+    def _column_types(self) -> tuple[str, str, str]:
+        # Return (payload_type, metadata_type, timestamp_type)
+        return "JSONB", "JSONB", "TIMESTAMPTZ"  # PostgreSQL
+
+    def _build_create_table_sql(self) -> str:
+        # Override for database-specific DDL syntax
+        return super()._build_create_table_sql()
+
+    def _wrap_create_statement(self, statement: str, object_type: str) -> str:
+        # Wrap with IF NOT EXISTS, PL/SQL blocks, etc.
+        return statement.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)
+```
+
+**Backend factory pattern** for native backends:
+
+```python
+# In sqlspec/adapters/{adapter}/events/backend.py
+def create_event_backend(
+    config: "AdapterConfig", backend_name: str, extension_settings: dict[str, Any]
+) -> AdapterEventsBackend | None:
+    if backend_name == "listen_notify":
+        return AdapterEventsBackend(config)
+    if backend_name == "listen_notify_durable":
+        queue = TableEventQueue(config, **extension_settings)
+        return AdapterHybridEventsBackend(config, QueueEventBackend(queue))
+    return None  # Falls back to table_queue
+```
+
+**Key principles:**
+
+- Backends implement `publish_async`, `dequeue_async`, `ack_async` (and sync variants)
+- Stores inherit from `BaseEventQueueStore` and override `_column_types()`
+- Use `driver_features` for backend selection, `extension_config["events"]` for settings
+- Always support `table_queue` fallback for databases without native pub/sub
+- Separate DDL execution for databases without transactional DDL (Spanner, BigQuery)
+
+**Reference implementation:** `sqlspec/extensions/events/`, `sqlspec/adapters/*/events/`
+
 ## Collaboration Guidelines
 
 - Challenge suboptimal requests constructively

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

@@ -14,7 +14,7 @@ from sqlspec.adapters.sqlite import SqliteConfig
 
 spec = SQLSpec()
 config = SqliteConfig(
-    pool_config={"database": ":memory:"},
+    connection_config={"database": ":memory:"},
     migration_config={
         "script_location": "migrations",
         "include_extensions": ["events"],
@@ -37,16 +37,17 @@ for message in channel.iter_events("notifications"):
 
 ## PostgreSQL native notifications
 
-Async PostgreSQL adapters (AsyncPG today, with psycopg/psqlpy on deck) default
-to `events_backend="listen_notify"`, which means no migrations are required;
-events flow directly through `LISTEN/NOTIFY`.
+All async PostgreSQL adapters (AsyncPG, Psycopg async, and Psqlpy) support
+native `LISTEN/NOTIFY` via `events_backend="listen_notify"`. When enabled,
+events flow directly through PostgreSQL's notification system with no
+migrations required.
 
 ```python
 from sqlspec import SQLSpec
 from sqlspec.adapters.asyncpg import AsyncpgConfig
 
 spec = SQLSpec()
-config = AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/db"})
+config = AsyncpgConfig(connection_config={"dsn": "postgresql://localhost/db"})
 spec.add_config(config)
 channel = spec.event_channel(config)
 
@@ -217,6 +218,7 @@ against accidental sync usage.
 | 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. |
 | BigQuery / ADBC | Queue fallback | `2.0s` | `60s` | Coarser cadence avoids hammering remote warehouses; still safe to override. |
+| Spanner | Queue fallback | `1.0s` | `30s` | Uses Spanner-native JSON and TIMESTAMP types; requires separate DDL execution. |
 | SQLite / AioSQLite | Queue fallback | `1.0s` | `30s` | General-purpose defaults that suit most local deployments. |
 
 ## Telemetry & observability
@@ -231,6 +233,84 @@ exports structured traces. The counters and spans flow through
 `SQLSpec.telemetry_snapshot()` plus the Prometheus helper when
 `extension_config["prometheus"]` is enabled.
 
+## Architecture
+
+The events extension consists of two layers that work together:
+
+### Event Backends
+
+Backends handle the actual pub/sub communication mechanism. SQLSpec supports
+three backend types:
+
+| Backend | Description | When to use |
+| --- | --- | --- |
+| `listen_notify` | Native PostgreSQL LISTEN/NOTIFY | Real-time, fire-and-forget events |
+| `listen_notify_durable` | Hybrid: queue table + NOTIFY wakeups | Real-time with durability and retries |
+| `advanced_queue` | Oracle Advanced Queuing | Enterprise Oracle deployments |
+| `table_queue` | Polling-based queue table | Universal fallback for any database |
+
+Configure the backend via `driver_features["events_backend"]`:
+
+```python
+from sqlspec.adapters.asyncpg import AsyncpgConfig
+
+# Native LISTEN/NOTIFY (default for PostgreSQL adapters)
+config = AsyncpgConfig(
+    connection_config={"dsn": "postgresql://localhost/db"},
+    driver_features={"events_backend": "listen_notify"},
+)
+
+# Hybrid: durable queue with NOTIFY wakeups
+config = AsyncpgConfig(
+    connection_config={"dsn": "postgresql://localhost/db"},
+    driver_features={"events_backend": "listen_notify_durable"},
+)
+```
+
+### Event Queue Stores
+
+Stores generate adapter-specific DDL for the queue table. Each adapter has
+a store class in `sqlspec/adapters/{adapter}/events/store.py` that handles:
+
+- Column type mapping (JSON, JSONB, CLOB, etc.)
+- Timestamp types (TIMESTAMPTZ, DATETIME, TIMESTAMP)
+- Index creation strategies
+- Database-specific DDL wrapping (IF NOT EXISTS, PL/SQL blocks, etc.)
+
+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 |
+| DuckDB | JSON | TIMESTAMP | Short poll intervals |
+| BigQuery | JSON | TIMESTAMP | CLUSTER BY for partitioning |
+| Spanner | JSON | TIMESTAMP | Separate DDL execution (no IF NOT EXISTS) |
+| ADBC | Dialect-detected | Dialect-detected | Multi-dialect support based on connection URI |
+| SQLite / AioSQLite | TEXT | TIMESTAMP | General-purpose defaults |
+
+### ADBC Multi-Dialect Support
+
+The ADBC adapter auto-detects the underlying database from the connection URI
+and generates appropriate DDL:
+
+```python
+from sqlspec.adapters.adbc import AdbcConfig
+
+# Connects to PostgreSQL via ADBC - uses JSONB columns
+config = AdbcConfig(
+    connection_config={"uri": "postgresql://localhost/db"},
+    extension_config={"events": {}},
+)
+
+# Connects to BigQuery via ADBC - uses JSON with CLUSTER BY
+config = AdbcConfig(
+    connection_config={"driver_name": "bigquery"},
+    extension_config={"events": {}},
+)
+```
+
 ## Litestar/ADK integration
 
 - The Litestar and ADK plugins already call `SQLSpec.event_channel()` for you.
@@ -246,7 +326,7 @@ from sqlspec.extensions.litestar import SQLSpecPlugin
 sql = SQLSpec()
 config = sql.add_config(
     AsyncpgConfig(
-        pool_config={"dsn": "postgresql://localhost/db"},
+        connection_config={"dsn": "postgresql://localhost/db"},
         extension_config={"events": {"queue_table": "app_events"}},
     )
 )

sqlspec/adapters/adbc/data_dictionary.py

@@ -216,14 +216,23 @@ def _get_dialect(self, driver: SyncDriverAdapterBase) -> str:
     def get_version(self, driver: SyncDriverAdapterBase) -> "VersionInfo | None":
         """Get database version information based on detected dialect.
 
+        Uses caching to avoid repeated database queries within the same
+        driver session.
+
         Args:
-            driver: ADBC driver instance
+            driver: ADBC driver instance.
 
         Returns:
-            Database version information or None if detection fails
+            Database version information or None if detection fails.
         """
+        driver_id = id(driver)
+        was_cached, cached_version = self.get_cached_version(driver_id)
+        if was_cached:
+            return cached_version
+
         dialect = self._get_dialect(driver)
         adbc_driver = cast("AdbcDriver", driver)
+        version_info: VersionInfo | None = None
 
         try:
             if dialect == "postgres":
@@ -234,39 +243,40 @@ def get_version(self, driver: SyncDriverAdapterBase) -> "VersionInfo | None":
                         major = int(match.group(1))
                         minor = int(match.group(2))
                         patch = int(match.group(3)) if match.group(3) else 0
-                        return VersionInfo(major, minor, patch)
+                        version_info = VersionInfo(major, minor, patch)
 
             elif dialect == "sqlite":
                 version_str = adbc_driver.select_value("SELECT sqlite_version()")
                 if version_str:
                     match = SQLITE_VERSION_PATTERN.match(str(version_str))
                     if match:
                         major, minor, patch = map(int, match.groups())
-                        return VersionInfo(major, minor, patch)
+                        version_info = VersionInfo(major, minor, patch)
 
             elif dialect == "duckdb":
                 version_str = adbc_driver.select_value("SELECT version()")
                 if version_str:
                     match = DUCKDB_VERSION_PATTERN.search(str(version_str))
                     if match:
                         major, minor, patch = map(int, match.groups())
-                        return VersionInfo(major, minor, patch)
+                        version_info = VersionInfo(major, minor, patch)
 
             elif dialect == "mysql":
                 version_str = adbc_driver.select_value("SELECT VERSION()")
                 if version_str:
                     match = MYSQL_VERSION_PATTERN.search(str(version_str))
                     if match:
                         major, minor, patch = map(int, match.groups())
-                        return VersionInfo(major, minor, patch)
+                        version_info = VersionInfo(major, minor, patch)
 
             elif dialect == "bigquery":
-                return VersionInfo(1, 0, 0)
+                version_info = VersionInfo(1, 0, 0)
 
         except Exception:
             logger.warning("Failed to get %s version", dialect)
 
-        return None
+        self.cache_version(driver_id, version_info)
+        return version_info
 
     def get_feature_flag(self, driver: SyncDriverAdapterBase, feature: str) -> bool:
         """Check if database supports a specific feature based on detected dialect.