SQLSpec Storage Backend¶

The SQLSpec backend (SQLSpecQueueBackend, configured via SQLSpecBackendConfig) provides high-throughput relational storage across 11 SQL dialects without requiring an ORM.

Runnable Example¶

"""Configuring SQLSpecBackendConfig with AiosqliteConfig, schema creation, and terminal_ttl."""

from pathlib import Path
from tempfile import TemporaryDirectory

from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues import QueueConfig, QueueService, WorkerConfig, task
from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecQueueBackend, SQLSpecWorkerWakeupConfig


@task("reports.archive_monthly", queue="reports")
async def archive_monthly(month: str) -> str:
    """Archive a monthly report record in SQLSpec storage."""
    return f"archived:{month}"


async def main() -> None:
    """Create the SQLSpec schema in a temporary SQLite database and execute a task."""
    with TemporaryDirectory() as tmp_dir:
        db_path = Path(tmp_dir) / "queues.sqlite3"
        sqlspec_config = AiosqliteConfig(connection_config={"database": str(db_path)})
        backend_config = SQLSpecBackendConfig(
            sqlspec_config=sqlspec_config,
            manage_schema=True,
            terminal_ttl=86_400.0,
            worker_wakeups=SQLSpecWorkerWakeupConfig(transport="polling"),
        )
        config = QueueConfig(
            queue_backend=backend_config, execution_backend="immediate", worker=WorkerConfig(placement="external")
        )
        async with QueueService(config=config) as service:
            backend = service.get_queue_backend()
            assert isinstance(backend, SQLSpecQueueBackend)
            await backend.create_schema()

            handle = await service.enqueue(archive_monthly, "2026-11")
            await handle.refresh()
            assert handle.status == "completed"
            assert handle.result == "archived:2026-11"

Supported SQLSpec Adapters & Dialects¶

litestar-queues ships dedicated SQLSpecQueueStore implementations tuned for each driver's parameter binding, locking (FOR UPDATE SKIP LOCKED), UPDATE ... RETURNING batch claiming, and JSON codecs:

Schema Management & Migrations¶

The SQLSpec queue schema (0001_create_queue_tasks) provisions four tables:

1. queue_task (override via queue_table_name): Active and terminal task records, payload JSON, lease timestamps, and retry fences.
2. queue_task_event_history (override via event_history_table_name): Durable lifecycle and custom progress events when QueueEventsConfig(history=EventHistoryConfig(...)) is active.
3. queue_maintenance (override via maintenance_table_name): Distributed maintenance lease coordination rows.
4. queue_task_reservation (override via task_reservation_table_name): Permanent unique_until="forever" identity reservations (TaskReservation).

Automatic vs Externally Managed Schemas¶

• SQLSpec Migration Integration: When SQLSpecBackendConfig is passed to QueueConfig, QueuePlugin invokes SQLSpecBackendConfig.configure_migrations() (configure_queue_migration_extension and configure_events_migration_extension) so sqlspec CLI migrations automatically include the queue and event-queue migration directories.
• Programmatic Schema Creation (create_schema()): Call await backend.create_schema() in tests, bootstrap scripts, or Google Cloud Spanner administrative DDL setup when manage_schema=True (the default).
• Adopter-Managed Schemas (manage_schema=False): Set manage_schema=False when your organization manages all DDL through external migration pipelines. When manage_schema=False, litestar-queues never executes CREATE TABLE or DROP TABLE statements and supports custom column mappings via column_map and native_json_columns.

Native Row-Level Terminal TTL & Worker Sweeps¶

Setting SQLSpecBackendConfig(terminal_ttl=86_400.0) configures retention for finished records (completed, failed, cancelled, expired) in two complementary ways:

1. Database-Native Row Expiration (CockroachDB & Google Cloud Spanner):
2. CockroachDB: create_statements() and migration 0001_create_queue_tasks attach a ttl_expiration_expression storage parameter computed from completed_at + INTERVAL '<N> seconds' so only rows with a non-null completed_at in terminal status expire automatically in the background.
3. Google Cloud Spanner: create_statements() attaches ROW DELETION POLICY (OLDER_THAN(completed_at, INTERVAL <N> DAY)) (rounding terminal_ttl up to whole days) because Spanner only deletes rows where completed_at IS NOT NULL.
4. In-Worker Periodic Cleanup (All SQL Dialects):
5. When WorkerConfig.terminal_retention is left as None, workers automatically inherit SQLSpecBackendConfig.terminal_ttl and run periodic cleanup_terminal() sweeps every WorkerConfig.terminal_check_interval (300.0s).

Worker Wakeup Transports¶

Configure SQLSpecWorkerWakeupConfig on SQLSpecBackendConfig.worker_wakeups to control how workers learn about newly enqueued tasks:

Passing worker_wakeups=None on SQLSpecBackendConfig is equivalent to SQLSpecWorkerWakeupConfig(transport="polling"). When notify_queue or poll_queue is selected, SQLSpecQueueBackend.open() validates at startup that the event-queue table exists and raises a descriptive QueueConfigurationError if migrations have not been applied.

Dedicated Heartbeat Connection Pool¶

Under heavy concurrency, long-running query transactions can contend with periodic worker heartbeat writes. Pass a separate SQLSpecStoreConfig with a small dedicated pool to SQLSpecBackendConfig.heartbeat_pool_config to isolate touch_heartbeats() traffic from normal enqueue and claim queries:

from litestar_queues.backends.sqlspec import SQLSpecBackendConfig
from sqlspec.adapters.asyncpg import AsyncpgConfig

main_pool = AsyncpgConfig(connection_config={"dsn": "postgresql://app:secret@db:5432/app", "min_size": 4, "max_size": 20})
heartbeat_pool = AsyncpgConfig(connection_config={"dsn": "postgresql://app:secret@db:5432/app", "min_size": 1, "max_size": 2})

backend_config = SQLSpecBackendConfig(
    sqlspec_config=main_pool,
    heartbeat_pool_config=heartbeat_pool,
)

Third-Party SQLSpec Adapter & Store Resolution¶

When using an external or custom SQLSpec adapter package outside sqlspec.adapters.*, litestar-queues resolves the adapter name and store classes through class attributes or sqlspec.utils.adapters helpers (resolve_adapter_name and resolve_extension_store_class) without requiring changes to internal adapter registries:

• adapter_name: ClassVar[str]: Declares the canonical adapter identifier on the custom SQLSpec config class (or resolves it from the base class MRO).
• queue_store_class: ClassVar[type[SQLSpecQueueStore]]: Points create_queue_store() directly to your custom SQLSpecQueueStore subclass.
• events_store_class: ClassVar[type[BaseEventQueueStore]]: Points SQLSpec's durable event channel to your custom event-queue store class when using notify_queue or poll_queue.

"""Declaring adapter_name and queue_store_class on a third-party SQLSpec adapter configuration."""

from typing import ClassVar

from sqlspec.adapters.aiosqlite import AiosqliteConfig

from litestar_queues.backends.sqlspec import SQLSpecBackendConfig, SQLSpecQueueBackend
from litestar_queues.backends.sqlspec.stores.aiosqlite import AiosqliteQueueStore
from litestar_queues.backends.sqlspec.stores.base import SQLSpecQueueStore
from litestar_queues.backends.sqlspec.stores.factory import create_queue_store


class CustomSqliteQueueStore(AiosqliteQueueStore):
    """Custom queue store subclass for a third-party or wrapped SQLite adapter."""

    __slots__ = ()


class CustomSqliteAdapterConfig(AiosqliteConfig):
    """Third-party SQLSpec adapter configuration advertising explicit store hooks."""

    adapter_name: ClassVar[str] = "custom_sqlite"
    queue_store_class: ClassVar[type[SQLSpecQueueStore]] = CustomSqliteQueueStore


async def main() -> None:
    """Verify create_queue_store and SQLSpecQueueBackend resolve queue_store_class directly."""
    adapter_config = CustomSqliteAdapterConfig(connection_config={"database": ":memory:"})
    store = create_queue_store(adapter_config, terminal_ttl=3600.0)
    assert isinstance(store, CustomSqliteQueueStore)

    backend = SQLSpecQueueBackend(
        backend_config=SQLSpecBackendConfig(sqlspec_config=adapter_config, terminal_ttl=3600.0)
    )
    assert isinstance(backend, SQLSpecQueueBackend)