API Reference

Complete API reference for the SQLSpec Litestar extension.

SQLSpecPlugin

class sqlspec.extensions.litestar.SQLSpecPlugin[source]

Bases: InitPluginProtocol, CLIPlugin

Litestar plugin for SQLSpec database integration.

Automatically configures NumPy array serialization when NumPy is installed, enabling seamless bidirectional conversion between NumPy arrays and JSON for vector embedding workflows.

Session Table Migrations:

The Litestar extension includes migrations for creating session storage tables. To include these migrations in your database migration workflow, add ‘litestar’ to the include_extensions list in your migration configuration.

Example

config = AsyncpgConfig(

pool_config={“dsn”: “postgresql://localhost/db”}, extension_config={

“litestar”: {

“session_table”: “custom_sessions” # Optional custom table name

}

}, migration_config={

“script_location”: “migrations”, “include_extensions”: [“litestar”], # Simple string list only

}

)

The session table migration will automatically use the appropriate column types for your database dialect (JSONB for PostgreSQL, JSON for MySQL, TEXT for SQLite).

Extension migrations use the ext_litestar_ prefix (e.g., ext_litestar_0001) to prevent version conflicts with application migrations.

__init__(sqlspec, *, loader=None)[source]

Initialize SQLSpec plugin.

Parameters:

  • sqlspec (SQLSpec) – Pre-configured SQLSpec instance with registered database configs.

  • loader (SQLFileLoader | None) – Optional SQL file loader instance (SQLSpec may already have one).

property config: list[SyncDatabaseConfig[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')] | NoPoolSyncConfig[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')] | AsyncDatabaseConfig[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')] | NoPoolAsyncConfig[TypeAliasForwardRef('typing.Any'), TypeAliasForwardRef('typing.Any')]]

Return the plugin configurations.

Returns:

List of database configurations.

on_cli_init(cli)[source]

Configure CLI commands for SQLSpec database operations.

Parameters:

cli (Group) – The Click command group to add commands to.

Return type:

None

on_app_init(app_config)[source]

Configure Litestar application with SQLSpec database integration.

Automatically registers NumPy array serialization when NumPy is installed.

Parameters:

app_config (AppConfig) – The Litestar application configuration instance.

Return type:

AppConfig

Returns:

The updated application configuration instance.

get_annotations()[source]

Return the list of annotations.

Return type:

list[type[Union[SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any]]]]

Returns:

List of annotations.

get_annotation(key)[source]

Return the annotation for the given configuration.

Parameters:

key (Union[str, SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any], type[Union[SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any]]]]) – The configuration instance or key to lookup.

Raises:

KeyError – If no configuration is found for the given key.

Return type:

type[Union[SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any]]]

Returns:

The annotation for the configuration.

get_config(name)[source]

Get a configuration instance by name.

Parameters:

name – The configuration identifier.

Raises:

KeyError – If no configuration is found for the given name.

Returns:

The configuration instance for the specified name.

provide_request_session(key, state, scope)[source]

Provide a database session for the specified configuration key from request scope.

Parameters:

  • key (Union[str, TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]), type[Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]]]) – The configuration identifier (same as get_config).

  • state (State) – The Litestar application State object.

  • scope (Union[HTTPScope, WebSocketScope]) – The ASGI scope containing the request context.

Return type:

SyncDriverAdapterBase | AsyncDriverAdapterBase

Returns:

A driver session instance for the specified database configuration.

provide_sync_request_session(key, state, scope)[source]

Provide a sync database session for the specified configuration key from request scope.

Parameters:

  • key (Union[str, TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), type[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any])]]) – The sync configuration identifier.

  • state (State) – The Litestar application State object.

  • scope (Union[HTTPScope, WebSocketScope]) – The ASGI scope containing the request context.

Return type:

SyncDriverAdapterBase

Returns:

A sync driver session instance for the specified database configuration.

provide_async_request_session(key, state, scope)[source]

Provide an async database session for the specified configuration key from request scope.

Parameters:

  • key (Union[str, TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]), type[TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]]) – The async configuration identifier.

  • state (State) – The Litestar application State object.

  • scope (Union[HTTPScope, WebSocketScope]) – The ASGI scope containing the request context.

Return type:

AsyncDriverAdapterBase

Returns:

An async driver session instance for the specified database configuration.

provide_request_connection(key, state, scope)[source]

Provide a database connection for the specified configuration key from request scope.

Parameters:

  • key (Union[str, TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]), type[Union[TypeVar(SyncConfigT, bound= SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]), TypeVar(AsyncConfigT, bound= AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any])]]]) – The configuration identifier (same as get_config).

  • state (State) – The Litestar application State object.

  • scope (Union[HTTPScope, WebSocketScope]) – The ASGI scope containing the request context.

Return type:

typing.Any

Returns:

A database connection instance for the specified database configuration.

Configuration

Litestar Configuration

Configure the plugin via extension_config in database configuration:

from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    pool_config={"dsn": "postgresql://localhost/mydb"},
    extension_config={
        "litestar": {
            "connection_key": "db_connection",
            "pool_key": "db_pool",
            "session_key": "db_session",
            "commit_mode": "autocommit",
            "extra_commit_statuses": {201, 204},
            "extra_rollback_statuses": {409},
            "enable_correlation_middleware": True,
            "correlation_header": "x-correlation-id",
        }
    }
)

Configuration Options

Session Stores

AsyncpgStore

class sqlspec.adapters.asyncpg.litestar.AsyncpgStore[source]

Bases: BaseSQLSpecStore[AsyncpgConfig]

PostgreSQL session store using AsyncPG driver.

Implements server-side session storage for Litestar using PostgreSQL via the AsyncPG driver. Provides efficient session management with: - Native async PostgreSQL operations - UPSERT support using ON CONFLICT - Automatic expiration handling - Efficient cleanup of expired sessions

Parameters:

config (AsyncpgConfig) – AsyncpgConfig instance with extension_config[“litestar”] settings.

Example

from sqlspec.adapters.asyncpg import AsyncpgConfig from sqlspec.adapters.asyncpg.litestar.store import AsyncpgStore

config = AsyncpgConfig(

pool_config={“dsn”: “postgresql://…”}, extension_config={“litestar”: {“session_table”: “my_sessions”}}

) store = AsyncpgStore(config) await store.create_table()

__init__(config)[source]

Initialize AsyncPG session store.

Parameters:

config (AsyncpgConfig) – AsyncpgConfig instance.

Notes

Table name is read from config.extension_config[“litestar”][“session_table”].

async create_table()[source]

Create the session table if it doesn’t exist.

Return type:

None

async get(key, renew_for=None)[source]

Get a session value by key.

Parameters:

  • key (str) – Session ID to retrieve.

  • renew_for (int | timedelta | None) – If given, renew the expiry time for this duration.

Return type:

bytes | None

Returns:

Session data as bytes if found and not expired, None otherwise.

Notes

Uses CURRENT_TIMESTAMP instead of NOW() for SQL standard compliance. The query planner can use the partial index for expires_at > CURRENT_TIMESTAMP.

async set(key, value, expires_in=None)[source]

Store a session value.

Parameters:
Return type:

None

Notes

Uses EXCLUDED to reference the proposed insert values in ON CONFLICT. Updates updated_at timestamp on every write for audit trail.

async delete(key)[source]

Delete a session by key.

Parameters:

key (str) – Session ID to delete.

Return type:

None

async delete_all()[source]

Delete all sessions from the store.

Return type:

None

async exists(key)[source]

Check if a session key exists and is not expired.

Parameters:

key (str) – Session ID to check.

Return type:

bool

Returns:

True if the session exists and is not expired.

Notes

Uses CURRENT_TIMESTAMP for consistency with get() method.

async expires_in(key)[source]

Get the time in seconds until the session expires.

Parameters:

key (str) – Session ID to check.

Return type:

int | None

Returns:

Seconds until expiration, or None if no expiry or key doesn’t exist.

async delete_expired()[source]

Delete all expired sessions.

Return type:

int

Returns:

Number of sessions deleted.

Notes

Uses CURRENT_TIMESTAMP for consistency. For very large tables (10M+ rows), consider batching deletes to avoid holding locks too long.

AiosqliteStore

class sqlspec.adapters.aiosqlite.litestar.AiosqliteStore[source]

Bases: BaseSQLSpecStore[AiosqliteConfig]

SQLite session store using AioSQLite driver.

Implements server-side session storage for Litestar using SQLite via the AioSQLite driver. Provides efficient session management with: - Native async SQLite operations - INSERT OR REPLACE for UPSERT functionality - Automatic expiration handling - Efficient cleanup of expired sessions

Parameters:

config (AiosqliteConfig) – AiosqliteConfig instance.

Example

from sqlspec.adapters.aiosqlite import AiosqliteConfig from sqlspec.adapters.aiosqlite.litestar.store import AiosqliteStore

config = AiosqliteConfig(database=”:memory:”) store = AiosqliteStore(config) await store.create_table()

__init__(config)[source]

Initialize AioSQLite session store.

Parameters:

config (AiosqliteConfig) – AiosqliteConfig instance.

Notes

Table name is read from config.extension_config[“litestar”][“session_table”].

async create_table()[source]

Create the session table if it doesn’t exist.

Return type:

None

async get(key, renew_for=None)[source]

Get a session value by key.

Parameters:

  • key (str) – Session ID to retrieve.

  • renew_for (int | timedelta | None) – If given, renew the expiry time for this duration.

Return type:

bytes | None

Returns:

Session data as bytes if found and not expired, None otherwise.

async set(key, value, expires_in=None)[source]

Store a session value.

Parameters:
Return type:

None

Notes

Stores expires_at as Julian Day number (REAL) for optimal index usage.

async delete(key)[source]

Delete a session by key.

Parameters:

key (str) – Session ID to delete.

Return type:

None

async delete_all()[source]

Delete all sessions from the store.

Return type:

None

async exists(key)[source]

Check if a session key exists and is not expired.

Parameters:

key (str) – Session ID to check.

Return type:

bool

Returns:

True if the session exists and is not expired.

async expires_in(key)[source]

Get the time in seconds until the session expires.

Parameters:

key (str) – Session ID to check.

Return type:

int | None

Returns:

Seconds until expiration, or None if no expiry or key doesn’t exist.

async delete_expired()[source]

Delete all expired sessions.

Return type:

int

Returns:

Number of sessions deleted.

OracledbStore

Commit Modes

Manual Mode

Explicit transaction control:

  • No automatic commits or rollbacks

  • Use async with session.begin_transaction()

  • Full control over transaction boundaries

Autocommit Mode

Automatic commit based on HTTP status:

Commits on:

  • HTTP 200-299 (success)

  • Any status in extra_commit_statuses

Rolls back on:

  • HTTP 300+ (redirects and errors)

  • Any status in extra_rollback_statuses

  • Exceptions during request handling

Autocommit with Redirects

Commits on success and redirects:

Commits on:

  • HTTP 200-399 (success + redirects)

  • Any status in extra_commit_statuses

Rolls back on:

  • HTTP 400+ (errors)

  • Any status in extra_rollback_statuses

  • Exceptions during request handling

Type Aliases

Common type annotations for dependency injection:

from sqlspec.driver import AsyncDriverAdapterBase, SyncDriverAdapterBase

# Async drivers
AsyncDriverAdapterBase  # Base class for all async drivers

# Sync drivers
SyncDriverAdapterBase   # Base class for all sync drivers

# Specific driver types

Usage Examples

Basic Plugin Setup

from litestar import Litestar
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.extensions.litestar import SQLSpecPlugin

spec = SQLSpec()
db = spec.add_config(
    AsyncpgConfig(pool_config={"dsn": "postgresql://localhost/mydb"})
)

plugin = SQLSpecPlugin(sqlspec=spec)
app = Litestar(route_handlers=[...], plugins=[plugin])

Multi-Database Setup

from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.duckdb import DuckDBConfig
from sqlspec.extensions.litestar import SQLSpecPlugin
from litestar import Litestar

spec = SQLSpec()

primary = spec.add_config(
    AsyncpgConfig(
        pool_config={"dsn": "postgresql://localhost/primary"},
        extension_config={
            "litestar": {"session_key": "primary_session"}
        }
    )
)

analytics = spec.add_config(
    DuckDBConfig(
        extension_config={
            "litestar": {"session_key": "analytics_session"}
        }
    )
)

plugin = SQLSpecPlugin(sqlspec=spec)
app = Litestar(route_handlers=[...], plugins=[plugin])

Session Store Setup

from litestar import Litestar
from litestar.middleware.session.server_side import ServerSideSessionConfig
from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig
from sqlspec.adapters.asyncpg.litestar import AsyncpgStore
from sqlspec.extensions.litestar import SQLSpecPlugin

# Create SQLSpec instance
spec = SQLSpec()

# Add database configuration
config = spec.add_config(
    AsyncpgConfig(
        pool_config={"dsn": "postgresql://localhost/mydb"},
        extension_config={"litestar": {"session_table": "litestar_sessions"}},
    )
)

# Create session store
store = AsyncpgStore(config)

# Configure Litestar application
app = Litestar(
    plugins=[SQLSpecPlugin(sqlspec=spec)],
    middleware=[
        ServerSideSessionConfig(store=store).middleware
    ]
)

See Also