Source code for sqlspec.extensions.events._store

"""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._buffer import validate_queue_capacity
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",
        "listener_queue_capacity",
        "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