"""Base classes for adapter-specific event queue stores."""
from abc import ABC, abstractmethod
from typing import TYPE_CHECKING, Any, ClassVar, Generic, TypeVar, cast
from sqlspec.exceptions import ImproperConfigurationError
from sqlspec.extensions.events._names import normalize_event_channel_name, normalize_queue_table_name
from sqlspec.migrations.schema import SchemaEnsureResult, SchemaTarget, ensure_schema_async, ensure_schema_sync
if TYPE_CHECKING:
from sqlspec.config import DatabaseConfigProtocol
__all__ = ("BaseEventQueueStore", "normalize_event_channel_name", "normalize_queue_table_name")
ConfigT = TypeVar("ConfigT", bound="DatabaseConfigProtocol[Any, Any, Any]")
[docs]
class BaseEventQueueStore(ABC, Generic[ConfigT]):
"""Base class for adapter-specific event queue DDL generators.
This class provides a hook-based pattern for DDL generation. Adapters only
need to override `_column_types()` and optionally any hook methods for
dialect-specific 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)
For complex dialects (Oracle PL/SQL, BigQuery CLUSTER BY), adapters may
override `_table_ddl()` directly.
"""
__slots__ = ("_config", "_extension_settings", "_table_name")
extension_config_options: ClassVar[frozenset[str]] = frozenset({
"backend",
"create_schema",
"event_poll_interval",
"lease_seconds",
"manage_schema",
"migrations_path",
"poll_interval",
"queue_table",
"retention_seconds",
"run_migrations",
"select_for_update",
"skip_locked",
})
[docs]
def __init__(self, config: ConfigT) -> None:
self._config = config
extension_config = cast("dict[str, Any]", config.extension_config)
self._extension_settings = cast("dict[str, Any]", extension_config.get("events", {}))
self._validate_extension_config()
table_name = self._extension_settings.get("queue_table", "sqlspec_event_queue")
self._table_name = normalize_queue_table_name(str(table_name))
@property
def table_name(self) -> str:
"""Return the configured queue table name."""
return self._table_name
@property
def settings(self) -> "dict[str, Any]":
"""Return extension settings for adapters to inspect."""
return self._extension_settings
[docs]
def create_statements(self) -> "list[str]":
"""Return statements required to create the queue table and indexes."""
statements = [self._wrap_create_statement(self._table_ddl(), "table")]
index_statement = self._index_ddl()
if index_statement:
statements.append(self._wrap_create_statement(index_statement, "index"))
return statements
[docs]
def drop_statements(self) -> "list[str]":
"""Return statements required to drop queue artifacts."""
return [self._wrap_drop_statement(f"DROP TABLE {self.table_name}")]
[docs]
def prepare_schema_sync(self, driver: Any) -> None:
"""Prepare adapter-specific schema decisions with a synchronous driver."""
[docs]
async def prepare_schema_async(self, driver: Any) -> None:
"""Prepare adapter-specific schema decisions with an asynchronous driver."""
[docs]
def reconcile_schema_sync(self, driver: Any) -> SchemaEnsureResult:
"""Apply additive queue-table changes with a synchronous driver."""
manage_schema, create_schema = self._schema_management_flags()
if not manage_schema:
return ensure_schema_sync(driver, [], manage_schema=False)
return ensure_schema_sync(driver, [self._schema_target()], manage_schema=True, create_schema=create_schema)
[docs]
async def reconcile_schema_async(self, driver: Any) -> SchemaEnsureResult:
"""Apply additive queue-table changes with an asynchronous driver."""
manage_schema, create_schema = self._schema_management_flags()
if not manage_schema:
return await ensure_schema_async(driver, [], manage_schema=False)
return await ensure_schema_async(
driver, [self._schema_target()], manage_schema=True, create_schema=create_schema
)
def _schema_target(self) -> SchemaTarget:
"""Build a schema target from the canonical queue table DDL."""
statement_config = getattr(self._config, "statement_config", None)
dialect = getattr(statement_config, "dialect", None)
return SchemaTarget.from_ddl(self.table_name, self.create_statements()[0], dialect=dialect)
def _validate_extension_config(self) -> None:
"""Reject events options that this adapter store cannot honor."""
unsupported = sorted(set(self._extension_settings).difference(type(self).extension_config_options))
if unsupported:
adapter = type(self).__module__.split(".")[2]
keys = ", ".join(repr(key) for key in unsupported)
msg = f"Unsupported events configuration key(s) for {adapter}: {keys}"
raise ImproperConfigurationError(msg)
def _schema_management_flags(self) -> "tuple[bool, bool]":
"""Return automatic-management and missing-table creation flags."""
return bool(self.settings.get("manage_schema", True)), bool(self.settings.get("create_schema", True))
def _string_type(self, length: int) -> str:
"""Return string type syntax for the given length.
Override for dialects with different string type syntax.
Args:
length: Maximum string length.
Returns:
String type declaration.
"""
return f"VARCHAR({length})"
def _integer_type(self) -> str:
"""Return integer type syntax.
Override for dialects with different integer type syntax.
Returns:
Integer type declaration.
"""
return "INTEGER"
def _timestamp_default(self) -> str:
"""Return timestamp default expression.
Override for dialects requiring different default syntax.
Returns:
Default timestamp expression.
"""
return "CURRENT_TIMESTAMP"
def _primary_key_syntax(self) -> str:
"""Return inline PRIMARY KEY clause for table definition.
Override for dialects that require PRIMARY KEY at the end of CREATE TABLE
instead of on the column definition.
Returns:
Empty string for column-level PK, or " PRIMARY KEY (event_id)" for table-level.
"""
return ""
def _table_ddl(self) -> str:
"""Build CREATE TABLE SQL using hook methods.
Most adapters should NOT override this method. Instead, override the
hook methods (_string_type, _integer_type, _timestamp_default, etc.)
for dialect-specific variations.
Only override this method for complex dialects that require entirely
different DDL structure.
"""
payload_type, metadata_type, timestamp_type = self._column_types()
string_64 = self._string_type(64)
string_128 = self._string_type(128)
string_32 = self._string_type(32)
integer_type = self._integer_type()
ts_default = self._timestamp_default()
pk_inline = self._primary_key_syntax()
table_clause = self._table_clause()
pk_column = " PRIMARY KEY" if not pk_inline else ""
return f"CREATE TABLE {self.table_name} (event_id {string_64}{pk_column}, channel {string_128} NOT NULL, payload_json {payload_type} NOT NULL, metadata_json {metadata_type}, status {string_32} NOT NULL DEFAULT 'pending', available_at {timestamp_type} NOT NULL DEFAULT {ts_default}, lease_expires_at {timestamp_type}, attempts {integer_type} NOT NULL DEFAULT 0, created_at {timestamp_type} NOT NULL DEFAULT {ts_default}, acknowledged_at {timestamp_type}){pk_inline}{table_clause}"
def _index_ddl(self) -> str | None:
"""Build CREATE INDEX SQL for queue operations."""
index_name = self._index_name()
return f"CREATE INDEX {index_name} ON {self.table_name}(channel, status, available_at)"
def _table_clause(self) -> str:
"""Return additional table options clause.
Override for dialects that need options after the column definitions.
"""
return ""
def _index_name(self) -> str:
"""Return the index name for the queue table."""
return f"idx_{self.table_name.replace('.', '_')}_channel_status"
def _index_existence_target(self) -> "tuple[str | None, str] | None":
"""Return the ``(schema, table)`` target for a data-dictionary index check.
Adapters whose index DDL is self-idempotent (``CREATE INDEX IF NOT
EXISTS``) return ``None``: no external existence check is needed. Adapters
whose dialect lacks an idempotent index DDL (MySQL) return the schema and
table so the migration can consult ``driver.data_dictionary.get_indexes``
and skip the ``ADD INDEX`` statement when the index already exists.
"""
return None
def _wrap_create_statement(self, statement: str, object_type: str) -> str:
"""Wrap CREATE statement with IF NOT EXISTS.
Override for dialects that don't support IF NOT EXISTS.
"""
if object_type == "table":
return statement.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)
if object_type == "index":
return statement.replace("CREATE INDEX", "CREATE INDEX IF NOT EXISTS", 1)
return statement
def _wrap_drop_statement(self, statement: str) -> str:
"""Wrap DROP statement with IF EXISTS.
Override for dialects that don't support IF EXISTS.
"""
return statement.replace("DROP TABLE", "DROP TABLE IF EXISTS", 1)
@abstractmethod
def _column_types(self) -> "tuple[str, str, str]":
"""Return payload, metadata, and timestamp column types for the adapter.
Args:
None
Returns:
Tuple of (payload_type, metadata_type, timestamp_type).
"""
