feat(channels): implement SQLSpecChannelsBackend for Litestar integration and add related tests

Author: cofin

docs/guides/extensions/litestar.md

@@ -150,6 +150,39 @@ app = Litestar(
 
 Every adapter exposes a store class (e.g., `AsyncpgStore`, `AiosqliteStore`, `DuckdbStore`) in its `litestar` submodule. Each subclass inherits `BaseSQLSpecStore`, enforces consistent schema DDL, and provides utilities such as `delete_expired()`. Run `delete_expired()` periodically or use `litestar sessions delete-expired` from the CLI.
 
+## Channels Backend (Database Broker)
+
+Litestar ships with a Channels plugin for broadcasting event streams to subscribers (for example WebSocket clients). SQLSpec can act as a Channels backend by reusing the `events` extension (table queue or native backends where available).
+
+```python
+from litestar import Litestar
+from litestar.channels.plugin import ChannelsPlugin
+
+from sqlspec import SQLSpec
+from sqlspec.adapters.aiosqlite import AiosqliteConfig
+from sqlspec.extensions.events import EventChannel
+from sqlspec.extensions.litestar import SQLSpecChannelsBackend, SQLSpecPlugin
+
+sqlspec = SQLSpec()
+config = sqlspec.add_config(
+    AiosqliteConfig(
+        connection_config={"database": "app.db"},
+        migration_config={"script_location": "migrations", "include_extensions": ["events"]},
+        extension_config={"events": {}},
+    )
+)
+
+channels_backend = SQLSpecChannelsBackend(EventChannel(config), channel_prefix="litestar")
+channels_plugin = ChannelsPlugin(backend=channels_backend, channels=["notifications"])
+
+app = Litestar(plugins=[SQLSpecPlugin(sqlspec), channels_plugin])
+```
+
+Notes:
+
+- Run migrations with `include_extensions=["events"]` so the queue table exists.
+- Litestar channels may use arbitrary names; the backend maps them to database-safe event channel identifiers deterministically.
+
 ## CLI Integration
 
 The `SQLSpecPlugin` automatically registers the `db` command group with the Litestar CLI. No additional configuration is required.

sqlspec/adapters/asyncmy/events/store.py

@@ -55,6 +55,32 @@ def _column_types(self) -> tuple[str, str, str]:
         """
         return "JSON", "JSON", "DATETIME(6)"
 
+    def _build_create_table_sql(self) -> str:
+        """Build MySQL-specific CREATE TABLE SQL.
+
+        MySQL requires CURRENT_TIMESTAMP(6) for DATETIME(6) columns,
+        not just CURRENT_TIMESTAMP which is only valid for TIMESTAMP type.
+
+        Returns:
+            CREATE TABLE SQL statement with MySQL-specific defaults.
+        """
+        payload_type, metadata_type, timestamp_type = self._column_types()
+        table_clause = self._table_clause()
+        return (
+            f"CREATE TABLE {self.table_name} ("
+            "event_id VARCHAR(64) PRIMARY KEY,"
+            " channel VARCHAR(128) NOT NULL,"
+            f" payload_json {payload_type} NOT NULL,"
+            f" metadata_json {metadata_type},"
+            " status VARCHAR(32) NOT NULL DEFAULT 'pending',"
+            f" available_at {timestamp_type} NOT NULL DEFAULT CURRENT_TIMESTAMP(6),"
+            f" lease_expires_at {timestamp_type},"
+            " attempts INTEGER NOT NULL DEFAULT 0,"
+            f" created_at {timestamp_type} NOT NULL DEFAULT CURRENT_TIMESTAMP(6),"
+            f" acknowledged_at {timestamp_type}"
+            f") {table_clause}"
+        )
+
     def _build_index_sql(self) -> str | None:
         """Build MySQL conditional index creation SQL.
 

sqlspec/adapters/asyncpg/events/backend.py

@@ -77,6 +77,10 @@ async def ack_async(self, event_id: str) -> None:
         await self._queue.ack_async(event_id)
         self._runtime.increment_metric("events.ack")
 
+    async def nack_async(self, event_id: str) -> None:
+        await self._queue.nack_async(event_id)
+        self._runtime.increment_metric("events.nack")
+
     def publish_sync(self, *_: Any, **__: Any) -> str:
         msg = "publish_sync is not supported for async-only Postgres backend"
         raise ImproperConfigurationError(msg)

sqlspec/adapters/oracledb/events/store.py

@@ -72,6 +72,8 @@ class _OracleEventQueueStoreMixin:
     Slots are defined in the concrete subclasses to allow proper attribute assignment.
     """
 
+    __slots__ = ()
+
     if TYPE_CHECKING:
         _in_memory: bool
         _json_storage: "JSONStorageType | None"
@@ -85,20 +87,21 @@ def _init_oracle_settings(self) -> None:
         """Initialize Oracle-specific settings from extension config.
 
         Must be called from subclass __init__ after super().__init__().
+        Note: Attributes assigned here are defined in subclass __slots__.
         """
         events_config = self._extension_settings
-        self._in_memory = bool(events_config.get("in_memory", False))
-        self._oracle_version_info = None
+        self._in_memory = bool(events_config.get("in_memory", False))  # type: ignore[misc]
+        self._oracle_version_info = None  # type: ignore[misc]
 
         json_storage_override = events_config.get("json_storage")
         if json_storage_override == "json":
-            self._json_storage = JSONStorageType.JSON_NATIVE
+            self._json_storage = JSONStorageType.JSON_NATIVE  # type: ignore[misc]
         elif json_storage_override == "blob_json":
-            self._json_storage = JSONStorageType.BLOB_JSON
+            self._json_storage = JSONStorageType.BLOB_JSON  # type: ignore[misc]
         elif json_storage_override == "blob":
-            self._json_storage = JSONStorageType.BLOB_PLAIN
+            self._json_storage = JSONStorageType.BLOB_PLAIN  # type: ignore[misc]
         else:
-            self._json_storage = None
+            self._json_storage = None  # type: ignore[misc]
 
     def _column_types(self) -> "tuple[str, str, str]":
         """Return Oracle column types based on storage mode."""

sqlspec/adapters/psqlpy/events/backend.py

@@ -246,6 +246,10 @@ async def ack_async(self, event_id: str) -> None:
         await self._queue.ack_async(event_id)
         self._runtime.increment_metric("events.ack")
 
+    async def nack_async(self, event_id: str) -> None:
+        await self._queue.nack_async(event_id)
+        self._runtime.increment_metric("events.nack")
+
     def ack_sync(self, _event_id: str) -> None:
         msg = "ack_sync is not supported for async-only Psqlpy backend"
         raise ImproperConfigurationError(msg)

sqlspec/adapters/psycopg/events/backend.py

@@ -327,6 +327,24 @@ def ack_sync(self, event_id: str) -> None:
         self._queue.ack_sync(event_id)
         self._runtime.increment_metric("events.ack")
 
+    async def nack_async(self, event_id: str) -> None:
+        """Return an event to the durable queue for redelivery asynchronously.
+
+        Args:
+            event_id: The event identifier to return to the queue.
+        """
+        await self._queue.nack_async(event_id)
+        self._runtime.increment_metric("events.nack")
+
+    def nack_sync(self, event_id: str) -> None:
+        """Return an event to the durable queue for redelivery synchronously.
+
+        Args:
+            event_id: The event identifier to return to the queue.
+        """
+        self._queue.nack_sync(event_id)
+        self._runtime.increment_metric("events.nack")
+
     async def shutdown_async(self) -> None:
         """Shutdown the async listener and release resources."""
         if self._listen_connection_async_cm is not None:

sqlspec/extensions/events/_queue.py

@@ -54,6 +54,7 @@ class TableEventQueue:
         "_json_passthrough",
         "_lease_seconds",
         "_max_claim_attempts",
+        "_nack_sql",
         "_retention_seconds",
         "_runtime",
         "_select_for_update",
@@ -95,6 +96,7 @@ def __init__(
         self._select_sql = self._build_select_sql()
         self._claim_sql = self._build_claim_sql()
         self._ack_sql = self._build_ack_sql()
+        self._nack_sql = self._build_nack_sql()
         self._acked_cleanup_sql = self._build_cleanup_sql()
 
     @property
@@ -135,6 +137,9 @@ def _build_claim_sql(self) -> str:
     def _build_ack_sql(self) -> str:
         return f"UPDATE {self._table_name} SET status = :acked, acknowledged_at = :acked_at WHERE event_id = :event_id"
 
+    def _build_nack_sql(self) -> str:
+        return f"UPDATE {self._table_name} SET status = :pending, lease_expires_at = NULL WHERE event_id = :event_id"
+
     def _build_cleanup_sql(self) -> str:
         return f"DELETE FROM {self._table_name} WHERE status = :acked AND acknowledged_at IS NOT NULL AND acknowledged_at <= :cutoff"
 
@@ -260,6 +265,26 @@ def ack_sync(self, event_id: str) -> None:
         self._cleanup_sync(now)
         self._runtime.increment_metric("events.ack")
 
+    async def nack_async(self, event_id: str) -> None:
+        """Return an event to the queue for redelivery asynchronously.
+
+        Resets the event status to pending and clears the lease, allowing it
+        to be picked up by another consumer. The attempts counter remains
+        unchanged (was already incremented during claim).
+        """
+        await self._execute_async(self._nack_sql, {"pending": _PENDING_STATUS, "event_id": event_id})
+        self._runtime.increment_metric("events.nack")
+
+    def nack_sync(self, event_id: str) -> None:
+        """Return an event to the queue for redelivery synchronously.
+
+        Resets the event status to pending and clears the lease, allowing it
+        to be picked up by another consumer. The attempts counter remains
+        unchanged (was already incremented during claim).
+        """
+        self._execute_sync(self._nack_sql, {"pending": _PENDING_STATUS, "event_id": event_id})
+        self._runtime.increment_metric("events.nack")
+
     async def _cleanup_async(self, reference: "datetime") -> None:
         cutoff = reference - timedelta(seconds=self._retention_seconds)
         await self._execute_async(self._acked_cleanup_sql, {"acked": _ACKED_STATUS, "cutoff": cutoff})
@@ -415,6 +440,12 @@ async def ack_async(self, event_id: str) -> None:
     def ack_sync(self, event_id: str) -> None:
         self._queue.ack_sync(event_id)
 
+    async def nack_async(self, event_id: str) -> None:
+        await self._queue.nack_async(event_id)
+
+    def nack_sync(self, event_id: str) -> None:
+        self._queue.nack_sync(event_id)
+
     @staticmethod
     def _to_message(event: QueueEvent) -> EventMessage:
         return EventMessage(

sqlspec/extensions/events/channel.py

@@ -250,6 +250,17 @@ def listen(
         thread.start()
         return listener
 
+    def listen_sync(
+        self, channel: str, handler: "SyncEventHandler", *, poll_interval: float | None = None, auto_ack: bool = True
+    ) -> SyncEventListener:
+        """Start a sync listener thread.
+
+        This is an alias of :meth:`listen`, provided for API symmetry with
+        :meth:`listen_async`.
+        """
+
+        return self.listen(channel, handler, poll_interval=poll_interval, auto_ack=auto_ack)
+
     def stop_listener(self, listener_id: str) -> None:
         """Stop a running sync listener."""
 
@@ -259,6 +270,15 @@ def stop_listener(self, listener_id: str) -> None:
         listener.stop()
         self._runtime.increment_metric("events.listener.stop")
 
+    def stop_listener_sync(self, listener_id: str) -> None:
+        """Stop a running sync listener.
+
+        This is an alias of :meth:`stop_listener`, provided for API symmetry
+        with :meth:`stop_listener_async`.
+        """
+
+        self.stop_listener(listener_id)
+
     def _run_sync_listener(
         self,
         listener_id: str,
@@ -391,6 +411,51 @@ def ack_sync(self, event_id: str) -> None:
             raise
         self._end_event_span(span, result="acked")
 
+    async def nack_async(self, event_id: str) -> None:
+        """Return an event to the queue for redelivery asynchronously.
+
+        Resets the event status to pending, clearing the lease and allowing
+        it to be picked up by another consumer.
+        """
+        if not self._is_async:
+            msg = "nack_async requires an async configuration"
+            raise ImproperConfigurationError(msg)
+        nack_method = getattr(self._backend, "nack_async", None)
+        if nack_method is None:
+            msg = "Current events backend does not support nack"
+            raise ImproperConfigurationError(msg)
+        span = self._start_event_span("nack", mode="async")
+        try:
+            await nack_method(event_id)
+        except Exception as error:
+            self._end_event_span(span, error=error)
+            raise
+        self._end_event_span(span, result="nacked")
+
+    def nack_sync(self, event_id: str) -> None:
+        """Return an event to the queue for redelivery synchronously.
+
+        Resets the event status to pending, clearing the lease and allowing
+        it to be picked up by another consumer.
+        """
+        if self._is_async:
+            if self._should_bridge_sync_calls():
+                self._bridge_sync_call(self.nack_async, event_id)
+                return
+            msg = "nack_sync requires a sync configuration"
+            raise ImproperConfigurationError(msg)
+        nack_method = getattr(self._backend, "nack_sync", None)
+        if nack_method is None:
+            msg = "Current events backend does not support nack"
+            raise ImproperConfigurationError(msg)
+        span = self._start_event_span("nack", mode="sync")
+        try:
+            nack_method(event_id)
+        except Exception as error:
+            self._end_event_span(span, error=error)
+            raise
+        self._end_event_span(span, result="nacked")
+
     # Loading helpers -----------------------------------------------------------
 
     @staticmethod
@@ -520,23 +585,18 @@ def _normalize_channel_name(channel: str) -> str:
             raise ImproperConfigurationError(str(error)) from error
 
     def _start_event_span(self, operation: str, channel: "str | None" = None, mode: str = "sync") -> Any:
-        span_manager = getattr(self._runtime, "span_manager", None)
-        if span_manager is None or not getattr(span_manager, "is_enabled", False):
+        if not getattr(self._runtime.span_manager, "is_enabled", False):
             return None
         attributes: dict[str, Any] = {
             "sqlspec.events.operation": operation,
             "sqlspec.events.backend": self._backend_name,
             "sqlspec.events.mode": mode,
-            "sqlspec.config": type(self._config).__name__,
         }
         if self._adapter_name:
             attributes["sqlspec.events.adapter"] = self._adapter_name
-        bind_key = getattr(self._config, "bind_key", None)
-        if bind_key:
-            attributes["sqlspec.bind_key"] = bind_key
         if channel:
             attributes["sqlspec.events.channel"] = channel
-        return span_manager.start_span(f"sqlspec.events.{operation}", attributes)
+        return self._runtime.start_span(f"sqlspec.events.{operation}", attributes=attributes)
 
     def _end_event_span(self, span: Any, *, error: "Exception | None" = None, result: "str | None" = None) -> None:
         if span is None:
@@ -545,4 +605,55 @@ def _end_event_span(self, span: Any, *, error: "Exception | None" = None, result
             setter = getattr(span, "set_attribute", None)
             if setter is not None:
                 setter("sqlspec.events.result", result)
-        self._runtime.span_manager.end_span(span, error=error)
+        self._runtime.end_span(span, error=error)
+
+    # Shutdown helpers ------------------------------------------------------------
+
+    async def shutdown_async(self) -> None:
+        """Shutdown the event channel and release backend resources.
+
+        Stops all async listeners and calls the backend's shutdown_async method
+        to release any held connections (e.g., LISTEN/NOTIFY listener connections).
+
+        Must be called before closing the database pool when using native backends.
+        """
+        span = self._start_event_span("shutdown", mode="async")
+        try:
+            for listener_id in list(self._listeners_async):
+                await self.stop_listener_async(listener_id)
+
+            backend_shutdown = getattr(self._backend, "shutdown_async", None)
+            if backend_shutdown is not None and callable(backend_shutdown):
+                result = backend_shutdown()
+                if result is not None:
+                    await result  # type: ignore[misc]
+        except Exception as error:
+            self._end_event_span(span, error=error)
+            raise
+        self._end_event_span(span, result="shutdown")
+        self._runtime.increment_metric("events.shutdown")
+
+    def shutdown_sync(self) -> None:
+        """Shutdown the event channel and release backend resources.
+
+        Stops all sync listeners. For async backends with sync bridging,
+        use shutdown_async instead.
+        """
+        if self._is_async:
+            if self._should_bridge_sync_calls():
+                self._bridge_sync_call(self.shutdown_async)
+                return
+            msg = "shutdown_sync requires a sync configuration"
+            raise ImproperConfigurationError(msg)
+        span = self._start_event_span("shutdown", mode="sync")
+        try:
+            for listener_id in list(self._listeners_sync):
+                self.stop_listener(listener_id)
+            backend_shutdown = getattr(self._backend, "shutdown_sync", None)
+            if backend_shutdown is not None and callable(backend_shutdown):
+                backend_shutdown()
+        except Exception as error:
+            self._end_event_span(span, error=error)
+            raise
+        self._end_event_span(span, result="shutdown")
+        self._runtime.increment_metric("events.shutdown")