feat: simplify event channel interfaces

- Updated event channel methods to use unified naming conventions for publish, ack, and iteration methods across async and sync configurations.
- Replaced `*_async` methods with their synchronous counterparts in tests and implementations.
- Adjusted test cases to reflect the new method names and ensure compatibility with the updated event channel structure.
- Enhanced type checking in tests by disabling specific pyright warnings where necessary.
- Removed deprecated portal bridge tests and ensured proper error handling for async/sync configuration mismatches.

Author: cofin

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

@@ -1,9 +1,9 @@
 # Database Event Channels
 
-SQLSpec now ships a portable event channel API that wraps native LISTEN/NOTIFY
+SQLSpec ships a portable event channel API that wraps native LISTEN/NOTIFY
 (or tapers down to a durable queue table when a driver lacks native support).
 This guide documents the queue-backed fallback delivered in
-`sqlspec.extensions.events.EventChannel`, the native PostgreSQL backend, and how to enable
+`sqlspec.extensions.events`, the native PostgreSQL backend, and how to enable
 each option across adapters.
 
 ## Quick start
@@ -51,14 +51,14 @@ config = AsyncpgConfig(connection_config={"dsn": "postgresql://localhost/db"})
 spec.add_config(config)
 channel = spec.event_channel(config)
 
-event_id = await channel.publish_async("notifications", {"type": "native"})
-async for message in channel.iter_events_async("notifications"):
+event_id = await channel.publish("notifications", {"type": "native"})
+async for message in channel.iter_events("notifications"):
     assert message.event_id == event_id
     break
 ```
 
 Native events are fire-and-forget: there is no durable queue row, so
-`ack_async()` becomes a no-op used solely for API parity. If you need durable
+`ack()` becomes a no-op used solely for API parity. If you need durable
 storage (e.g., for retries or multi-consumer fan-out) keep the queue backend
 enabled as described below.
 
@@ -117,14 +117,26 @@ When `in_memory=True` and the adapter dialect is Oracle, the migration adds the
 `INMEMORY PRIORITY HIGH` clause so queue rows live in the column store. Other
 adapters ignore this flag.
 
+## Sync vs Async Channels
+
+SQLSpec provides separate channel classes for sync and async configurations:
+
+- **`SyncEventChannel`** - For sync database configs (SqliteConfig, DuckdbConfig, PsycopgSyncConfig, etc.)
+- **`AsyncEventChannel`** - For async database configs (AsyncpgConfig, AiosqliteConfig, PsycopgAsyncConfig, etc.)
+
+The `SQLSpec.event_channel()` factory method automatically returns the correct type
+based on your configuration's `is_async` attribute.
+
 ## Publishing events
 
-Both sync and async adapters share the same API surface. Call the method that
-matches your driver type:
+Both sync and async channels share the same API surface with clean method names:
 
 ```python
+# Sync channel (SqliteConfig, DuckdbConfig, PsycopgSyncConfig, etc.)
 channel.publish("notifications", {"type": "user_update", "user_id": 42})
-await channel.publish_async("notifications", {"type": "refresh", "user_id": 42})
+
+# Async channel (AsyncpgConfig, AiosqliteConfig, PsycopgAsyncConfig, etc.)
+await channel.publish("notifications", {"type": "refresh", "user_id": 42})
 ```
 
 Payloads must be JSON-serialisable. Optional metadata maps can be stored via the
@@ -145,23 +157,23 @@ async def handle(message: EventMessage) -> None:
     # do work, then ack when auto_ack=False
     print(message.channel, message.payload)
 
-listener = channel.listen_async(
+listener = channel.listen(
     "notifications",
     handle,
     poll_interval=0.5,
     auto_ack=True,
 )
 
 # later
-await channel.stop_listener_async(listener.id)
+await channel.stop_listener(listener.id)
 ```
 
 For manual iteration instead of background tasks:
 
 ```python
-async for message in channel.iter_events_async("notifications", poll_interval=1):
+async for message in channel.iter_events("notifications", poll_interval=1):
     await process(message)
-    await channel.ack_async(message.event_id)
+    await channel.ack(message.event_id)
 ```
 
 ### Sync listeners
@@ -184,16 +196,6 @@ channel.stop_listener(listener.id)
 Manual iteration is also available via `channel.iter_events(...)` which yields
 `EventMessage` objects until you break the loop.
 
-#### Using sync APIs with async adapters
-
-When you call `SQLSpec.event_channel()` with an async adapter (AsyncPG,
-AioSQLite, etc.) the extension automatically enables a *portal bridge* so the
-sync APIs (`publish`, `iter_events`, `listen`, `ack`) remain usable. Under the
-hood SQLSpec runs the async backend inside a background event loop via
-`sqlspec.utils.portal`. Disable this by setting
-`extension_config["events"]["portal_bridge"] = False` if you prefer to guard
-against accidental sync usage.
-
 ## Configuration reference
 
 | Option             | Default                 | Description |
@@ -244,21 +246,32 @@ The durable `table_queue` backend supports returning a claimed message to the
 queue for redelivery:
 
 ```python
-message = await channel.dequeue_async("notifications", poll_interval=1.0)
-if message is not None:
-    await channel.nack_async(message.event_id)
+# Async channel
+async for message in channel.iter_events("notifications", poll_interval=1.0):
+    await channel.nack(message.event_id)
+    break
+
+# Sync channel
+for message in channel.iter_events("notifications", poll_interval=1.0):
+    channel.nack(message.event_id)
+    break
 ```
 
 Native `listen_notify` backends are fire-and-forget and do not support nacking
 because they do not create a durable queue row.
 
 When using native PostgreSQL backends (or hybrid backends that keep a dedicated
-listener connection), call `shutdown_async()` before closing the pool to release
+listener connection), call `shutdown()` before closing the pool to release
 listener resources cleanly:
 
 ```python
-await channel.shutdown_async()
+# Async channel
+await channel.shutdown()
 await config.close_pool()
+
+# Sync channel
+channel.shutdown()
+config.close_pool()
 ```
 
 ## Architecture

sqlspec/__init__.py

@@ -39,7 +39,13 @@
 from sqlspec.core import filters as filters
 from sqlspec.driver import AsyncDriverAdapterBase, ExecutionResult, SyncDriverAdapterBase
 from sqlspec.exceptions import StackExecutionError
-from sqlspec.extensions.events import AsyncEventListener, EventChannel, EventMessage, SyncEventListener
+from sqlspec.extensions.events import (
+    AsyncEventChannel,
+    AsyncEventListener,
+    EventMessage,
+    SyncEventChannel,
+    SyncEventListener,
+)
 from sqlspec.loader import SQLFile, SQLFileLoader
 from sqlspec.typing import ConnectionT, PoolT, SchemaT, StatementParameters, SupportedSchemaModel
 from sqlspec.utils.logging import suppress_erroneous_sqlglot_log_messages
@@ -51,6 +57,7 @@
     "ArrowResult",
     "AsyncDatabaseConfig",
     "AsyncDriverAdapterBase",
+    "AsyncEventChannel",
     "AsyncEventListener",
     "CacheConfig",
     "CacheStats",
@@ -60,7 +67,6 @@
     "CreateTable",
     "Delete",
     "DropTable",
-    "EventChannel",
     "EventMessage",
     "ExecutionResult",
     "FunctionColumn",
@@ -90,6 +96,7 @@
     "SupportedSchemaModel",
     "SyncDatabaseConfig",
     "SyncDriverAdapterBase",
+    "SyncEventChannel",
     "SyncEventListener",
     "Update",
     "__version__",

sqlspec/adapters/asyncpg/events/backend.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import contextlib
-import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
@@ -13,6 +12,7 @@
 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
+from sqlspec.utils.uuids import uuid4
 
 if TYPE_CHECKING:
     from sqlspec.adapters.asyncpg.config import AsyncpgConfig
@@ -46,7 +46,7 @@ def __init__(self, config: "AsyncpgConfig", queue: "QueueEventBackend") -> None:
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         await self._publish_durable(channel, event_id, payload, metadata)
         self._runtime.increment_metric("events.publish.native")
         return event_id
@@ -183,7 +183,7 @@ def __init__(self, config: "AsyncpgConfig") -> None:
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         envelope = self._encode_payload(event_id, payload, metadata)
         async with self._config.provide_session() as driver:
             await driver.execute(SQL("SELECT pg_notify($1, $2)", channel, envelope))
@@ -314,7 +314,7 @@ def _decode_payload(self, channel: str, payload: str) -> "EventMessage":
         data = from_json(payload)
         if not isinstance(data, dict):
             data = {"payload": data}
-        event_id = data.get("event_id", uuid.uuid4().hex)
+        event_id = data.get("event_id", uuid4().hex)
         payload_obj = data.get("payload")
         if not isinstance(payload_obj, dict):
             payload_obj = {"value": payload_obj}

sqlspec/adapters/oracledb/events/backend.py

@@ -1,13 +1,13 @@
 """Oracle Advanced Queuing backend for EventChannel."""
 
 import contextlib
-import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
 from sqlspec.exceptions import EventChannelError, ImproperConfigurationError, MissingDependencyError
 from sqlspec.extensions.events._models import EventMessage
 from sqlspec.utils.logging import get_logger
+from sqlspec.utils.uuids import uuid4
 
 if TYPE_CHECKING:
     from sqlspec.config import DatabaseConfigProtocol
@@ -49,7 +49,7 @@ def __init__(self, config: "DatabaseConfigProtocol[Any, Any, Any]", settings: di
         self._wait_seconds: int = int(settings.get("aq_wait_seconds", 5))
 
     def publish_sync(self, channel: str, payload: dict[str, Any], metadata: dict[str, Any] | None = None) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         envelope = self._build_envelope(channel, event_id, payload, metadata)
         session_cm = self._config.provide_session()
         with session_cm as driver:  # type: ignore[union-attr]
@@ -97,7 +97,7 @@ def dequeue_sync(self, channel: str, poll_interval: float) -> EventMessage | Non
             payload = {"payload": payload}
         payload_channel = payload.get("channel")
         message_channel = payload_channel if isinstance(payload_channel, str) else channel
-        event_id = payload.get("event_id", uuid.uuid4().hex)
+        event_id = payload.get("event_id", uuid4().hex)
         body = payload.get("payload")
         if not isinstance(body, dict):
             body = {"value": body}

sqlspec/adapters/psqlpy/events/backend.py

@@ -2,7 +2,6 @@
 
 import asyncio
 import contextlib
-import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
@@ -12,6 +11,7 @@
 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
+from sqlspec.utils.uuids import uuid4
 
 if TYPE_CHECKING:
     from psqlpy import Listener
@@ -52,7 +52,7 @@ def __init__(self, config: "PsqlpyConfig") -> None:
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         envelope = self._encode_payload(event_id, payload, metadata)
         if len(envelope.encode("utf-8")) > MAX_NOTIFY_BYTES:
             msg = "PostgreSQL NOTIFY payload exceeds 8 KB limit"
@@ -144,7 +144,7 @@ def _decode_payload(channel: str, payload: str) -> EventMessage:
         data = from_json(payload)
         if not isinstance(data, dict):
             data = {"payload": data}
-        event_id = data.get("event_id", uuid.uuid4().hex)
+        event_id = data.get("event_id", uuid4().hex)
         payload_obj = data.get("payload")
         if not isinstance(payload_obj, dict):
             payload_obj = {"value": payload_obj}
@@ -208,7 +208,7 @@ def __init__(self, config: "PsqlpyConfig", queue_backend: "QueueEventBackend") -
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         await self._publish_durable_async(channel, event_id, payload, metadata)
         self._runtime.increment_metric("events.publish.native")
         return event_id

sqlspec/adapters/psycopg/events/backend.py

@@ -2,7 +2,6 @@
 """Psycopg LISTEN/NOTIFY and hybrid event backends."""
 
 import contextlib
-import uuid
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING, Any
 
@@ -13,6 +12,7 @@
 from sqlspec.extensions.events._store import normalize_event_channel_name
 from sqlspec.utils.logging import get_logger
 from sqlspec.utils.serializers import from_json, to_json
+from sqlspec.utils.uuids import uuid4
 
 if TYPE_CHECKING:
     from sqlspec.adapters.psycopg.config import PsycopgAsyncConfig, PsycopgSyncConfig
@@ -54,7 +54,7 @@ def __init__(self, config: "PsycopgAsyncConfig | PsycopgSyncConfig") -> None:
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         envelope = self._encode_payload(event_id, payload, metadata)
         session_cm = self._config.provide_session()
         async with session_cm as driver:  # type: ignore[union-attr]
@@ -64,7 +64,7 @@ async def publish_async(
         return event_id
 
     def publish_sync(self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         envelope = self._encode_payload(event_id, payload, metadata)
         session_cm = self._config.provide_session()
         with session_cm as driver:  # type: ignore[union-attr]
@@ -179,7 +179,7 @@ def _decode_payload(channel: str, payload: str) -> EventMessage:
         data = from_json(payload)
         if not isinstance(data, dict):
             data = {"payload": data}
-        event_id = data.get("event_id", uuid.uuid4().hex)
+        event_id = data.get("event_id", uuid4().hex)
         payload_obj = data.get("payload")
         if not isinstance(payload_obj, dict):
             payload_obj = {"value": payload_obj}
@@ -249,13 +249,13 @@ def __init__(self, config: "DatabaseConfigProtocol[Any, Any, Any]", queue_backen
     async def publish_async(
         self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None
     ) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         await self._publish_durable_async(channel, event_id, payload, metadata)
         self._runtime.increment_metric("events.publish.native")
         return event_id
 
     def publish_sync(self, channel: str, payload: "dict[str, Any]", metadata: "dict[str, Any] | None" = None) -> str:
-        event_id = uuid.uuid4().hex
+        event_id = uuid4().hex
         self._publish_durable_sync(channel, event_id, payload, metadata)
         self._runtime.increment_metric("events.publish.native")
         return event_id