feat: simplify event backend (#297)

Author: cofin

AGENTS.md

@@ -96,6 +96,35 @@ SQLSpec is a type-safe SQL query mapper designed for minimal abstraction between
 - **Single-Pass Processing**: Parse once → transform once → validate once - SQL object is single source of truth
 - **Abstract Methods with Concrete Implementations**: Protocol defines abstract methods, base classes provide concrete sync/async implementations
 
+### Adapter Transaction Detection Pattern
+
+Each adapter MUST override `_connection_in_transaction()` with direct attribute access instead of using the base class fallback which relies on `getattr()` chains.
+
+```python
+# In each adapter's driver.py
+class MyAdapterDriver(SyncDriverBase):
+    def _connection_in_transaction(self) -> bool:
+        # AsyncPG: uses is_in_transaction() method
+        return self.connection.is_in_transaction()
+
+        # SQLite/DuckDB: uses in_transaction property
+        return self.connection.in_transaction
+
+        # Psycopg: uses status attribute
+        return self.connection.status != psycopg.pq.TransactionStatus.IDLE
+
+        # BigQuery: No transaction support
+        return False
+```
+
+**Why this matters:**
+
+- The base class uses `getattr()` chains which are slow and prevent mypyc optimization
+- Each adapter knows exactly which attribute to check
+- Direct attribute access is 10-50x faster in hot paths
+
+**Reference implementations:** All adapters in `sqlspec/adapters/*/driver.py` have this override.
+
 ### Query Stack Implementation Guidelines
 
 - **Builder Discipline**
@@ -290,6 +319,60 @@ def test_starlette_autocommit_mode() -> None:
 - No inline comments - use docstrings
 - Google-style docstrings with Args, Returns, Raises sections
 
+### Type Guards Pattern
+
+When checking object capabilities, ALWAYS use type guards from `sqlspec.utils.type_guards` instead of `hasattr()`:
+
+```python
+# NEVER do this - breaks mypyc and is slow
+if hasattr(obj, "expression") and hasattr(obj, "sql"):
+    sql = obj.sql
+    expr = obj.expression
+
+# ALWAYS do this - type-safe and fast
+from sqlspec.utils.type_guards import has_expression_and_sql
+if has_expression_and_sql(obj):
+    sql = obj.sql  # Type checker knows these exist
+    expr = obj.expression
+```
+
+**Available type guards:**
+
+| Guard | Checks For | Use When |
+|-------|------------|----------|
+| `is_readable(obj)` | `.read()` method | Checking LOB/stream objects |
+| `has_array_interface(obj)` | `.dtype` and `.shape` | NumPy-like arrays |
+| `has_cursor_metadata(obj)` | `.description` | Cursor metadata access |
+| `has_expression_and_sql(obj)` | `.expression` and `.sql` | SQL statement objects |
+| `has_expression_and_parameters(obj)` | `.expression` and `.parameters` | Parameterized statements |
+| `is_statement_filter(obj)` | `.append_to_statement()` | Statement filter objects |
+
+**Creating new type guards:**
+
+1. Add protocol to `sqlspec/protocols.py`:
+
+```python
+class MyProtocol(Protocol):
+    """Protocol for objects with specific capability."""
+    my_attribute: str
+    def my_method(self) -> None: ...
+```
+
+2. Add type guard to `sqlspec/utils/type_guards.py`:
+
+```python
+def has_my_capability(obj: Any) -> TypeGuard[MyProtocol]:
+    """Check if object has my_attribute and my_method."""
+    return hasattr(obj, "my_attribute") and hasattr(obj, "my_method")
+```
+
+**Key principles:**
+
+- Type guards centralize `hasattr()` calls in ONE place
+- Protocols provide type narrowing after the guard passes
+- Type checkers understand the guard's implications
+- Works with mypyc compilation
+
 ### Testing
 
 - **MANDATORY**: Function-based tests only (`def test_something():`)
@@ -887,26 +970,37 @@ Available backends:
 PostgreSQL adapters (asyncpg, psycopg, psqlpy) default to `listen_notify`.
 All other adapters default to `table_queue`.
 
-**Stores** generate adapter-specific DDL for the queue table:
+**Stores** generate adapter-specific DDL using a hook-based pattern:
 
 ```python
 # In sqlspec/adapters/{adapter}/events/store.py
 class AdapterEventQueueStore(BaseEventQueueStore[AdapterConfig]):
     __slots__ = ()
 
+    # REQUIRED: Return (payload_type, metadata_type, timestamp_type)
     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()
+    # OPTIONAL hooks for dialect variations:
+    def _string_type(self, length: int) -> str:
+        return f"VARCHAR({length})"  # Override for STRING(N), VARCHAR2(N), etc.
 
-    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)
+    def _integer_type(self) -> str:
+        return "INTEGER"  # Override for INT64, NUMBER(10), etc.
+
+    def _timestamp_default(self) -> str:
+        return "CURRENT_TIMESTAMP"  # Override for CURRENT_TIMESTAMP(6), SYSTIMESTAMP, etc.
+
+    def _primary_key_syntax(self) -> str:
+        return ""  # Override for " PRIMARY KEY (event_id)" if PK must be inline
+
+    def _table_clause(self) -> str:
+        return ""  # Override for " CLUSTER BY ..." or " INMEMORY ..."
 ```
 
+Most adapters only override `_column_types()`. Complex dialects may override
+`_build_create_table_sql()` directly (Oracle PL/SQL, BigQuery CLUSTER BY, Spanner no-DEFAULT).
+
 **Backend factory pattern** for native backends:
 
 ```python

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

@@ -318,13 +318,23 @@ use `migration_config={"exclude_extensions": ["events"]}`.
 
 ### 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.)
+Stores generate adapter-specific DDL for the queue table using a hook-based
+pattern. The base class `BaseEventQueueStore` provides a template for DDL
+generation, and adapters override hook methods for dialect-specific variations:
+
+**Required hook:**
+- `_column_types()` - Return tuple of (payload_type, metadata_type, timestamp_type)
+
+**Optional hooks for dialect 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)
+
+Most adapters only need to override `_column_types()`. Complex dialects
+(Oracle PL/SQL, BigQuery CLUSTER BY, Spanner no-DEFAULT) may override
+`_build_create_table_sql()` directly.
 
 Example store implementations:
 

sqlspec/adapters/adbc/driver.py

@@ -605,6 +605,16 @@ def commit(self) -> None:
             msg = f"Failed to commit transaction: {e}"
             raise SQLSpecError(msg) from e
 
+    def _connection_in_transaction(self) -> bool:
+        """Check if connection is in transaction.
+
+        ADBC uses explicit BEGIN and does not expose reliable transaction state.
+
+        Returns:
+            False - ADBC requires explicit transaction management.
+        """
+        return False
+
     @property
     def data_dictionary(self) -> "SyncDataDictionaryBase":
         """Get the data dictionary for this driver.

sqlspec/adapters/aiosqlite/driver.py

@@ -390,6 +390,14 @@ async def _truncate_table_async(self, table: str) -> None:
         async with self.handle_database_exceptions(), self.with_cursor(self.connection) as cursor:
             await cursor.execute(statement)
 
+    def _connection_in_transaction(self) -> bool:
+        """Check if connection is in transaction.
+
+        Returns:
+            True if connection is in an active transaction.
+        """
+        return bool(self.connection.in_transaction)
+
     @property
     def data_dictionary(self) -> "AsyncDataDictionaryBase":
         """Get the data dictionary for this driver.

sqlspec/adapters/asyncmy/driver.py

@@ -550,6 +550,16 @@ async def _truncate_table_async(self, table: str) -> None:
         async with self.handle_database_exceptions(), self.with_cursor(self.connection) as cursor:
             await cursor.execute(statement)
 
+    def _connection_in_transaction(self) -> bool:
+        """Check if connection is in transaction.
+
+        AsyncMy uses explicit BEGIN and does not expose reliable transaction state.
+
+        Returns:
+            False - AsyncMy requires explicit transaction management.
+        """
+        return False
+
     @property
     def data_dictionary(self) -> "AsyncDataDictionaryBase":
         """Get the data dictionary for this driver.

sqlspec/adapters/asyncmy/events/store.py

@@ -55,31 +55,16 @@ 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.
+    def _timestamp_default(self) -> str:
+        """Return MySQL timestamp default expression.
 
         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.
+            MySQL-specific timestamp default with microsecond precision.
         """
-        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}"
-        )
+        return "CURRENT_TIMESTAMP(6)"
 
     def _build_index_sql(self) -> str | None:
         """Build MySQL conditional index creation SQL.

sqlspec/adapters/asyncpg/driver.py

@@ -347,47 +347,57 @@ async def _execute_stack_native(
     ) -> "tuple[StackResult, ...]":
         results: list[StackResult] = []
 
-        async def _run_operations(observer: StackExecutionObserver) -> None:
-            for index, operation in enumerate(stack.operations):
-                try:
-                    normalized = None
-                    if operation.method == "execute":
-                        normalized = self._normalize_stack_execute_operation(operation)
-
-                    if normalized is not None and self._can_prepare_stack_operation(normalized):
-                        stack_result = await self._execute_stack_operation_prepared(normalized)
-                    else:
-                        result = await self._execute_stack_operation(operation)
-                        stack_result = StackResult(result=result)
-                except Exception as exc:
-                    stack_error = StackExecutionError(
-                        index,
-                        describe_stack_statement(operation.statement),
-                        exc,
-                        adapter=type(self).__name__,
-                        mode="continue-on-error" if continue_on_error else "fail-fast",
-                    )
-                    if continue_on_error:
-                        observer.record_operation_error(stack_error)
-                        results.append(StackResult.from_error(stack_error))
-                        continue
-                    raise stack_error from exc
-
-                results.append(stack_result)
-
         transaction_cm = None
         if not continue_on_error and not self._connection_in_transaction():
             transaction_cm = self.connection.transaction()
 
         with StackExecutionObserver(self, stack, continue_on_error, native_pipeline=True) as observer:
             if transaction_cm is not None:
                 async with transaction_cm:
-                    await _run_operations(observer)
+                    await self._run_stack_operations(stack, continue_on_error, observer, results)
             else:
-                await _run_operations(observer)
+                await self._run_stack_operations(stack, continue_on_error, observer, results)
 
         return tuple(results)
 
+    async def _run_stack_operations(
+        self,
+        stack: "StatementStack",
+        continue_on_error: bool,
+        observer: "StackExecutionObserver",
+        results: "list[StackResult]",
+    ) -> None:
+        """Run operations for statement stack execution.
+
+        Extracted from _execute_stack_native to avoid closure compilation issues.
+        """
+        for index, operation in enumerate(stack.operations):
+            try:
+                normalized = None
+                if operation.method == "execute":
+                    normalized = self._normalize_stack_execute_operation(operation)
+
+                if normalized is not None and self._can_prepare_stack_operation(normalized):
+                    stack_result = await self._execute_stack_operation_prepared(normalized)
+                else:
+                    result = await self._execute_stack_operation(operation)
+                    stack_result = StackResult(result=result)
+            except Exception as exc:
+                stack_error = StackExecutionError(
+                    index,
+                    describe_stack_statement(operation.statement),
+                    exc,
+                    adapter=type(self).__name__,
+                    mode="continue-on-error" if continue_on_error else "fail-fast",
+                )
+                if continue_on_error:
+                    observer.record_operation_error(stack_error)
+                    results.append(StackResult.from_error(stack_error))
+                    continue
+                raise stack_error from exc
+
+            results.append(stack_result)
+
     async def _execute_statement(self, cursor: "AsyncpgConnection", statement: "SQL") -> "ExecutionResult":
         """Execute single SQL statement.
 
@@ -623,6 +633,10 @@ async def _truncate_table(self, table: str) -> None:
             msg = f"Failed to truncate table '{table}': {exc}"
             raise SQLSpecError(msg) from exc
 
+    def _connection_in_transaction(self) -> bool:
+        """Check if connection is in transaction."""
+        return bool(self.connection.is_in_transaction())
+
 
 def _convert_datetime_param(value: Any) -> Any:
     """Convert datetime parameter, handling ISO strings."""