Litestar

Full Litestar integration with plugin lifecycle, dependency injection, CLI commands, channels backend, and key-value store.

Plugin

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(

connection_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]
Overloads:

  • self, name (type[SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]]) → SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any]

  • self, name (type[AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]]) → AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]

  • self, name (str) → SyncDatabaseConfig[Any, Any, Any] | NoPoolSyncConfig[Any, Any] | AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, Any]

Get a configuration instance by name.

Parameters:

name (Union[type[DatabaseConfigProtocol[typing.Any, typing.Any, typing.Any]], str, typing.Any]) -- 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]
Overloads:

  • self, key (SyncDatabaseConfig[Any, Any, DriverT] | NoPoolSyncConfig[Any, DriverT] | type[SyncDatabaseConfig[Any, Any, DriverT] | NoPoolSyncConfig[Any, DriverT]]), state (State), scope (Scope) → DriverT

  • self, key (AsyncDatabaseConfig[Any, Any, DriverT] | NoPoolAsyncConfig[Any, DriverT] | type[AsyncDatabaseConfig[Any, Any, DriverT] | NoPoolAsyncConfig[Any, DriverT]]), state (State), scope (Scope) → DriverT

  • self, key (str), state (State), scope (Scope) → SyncDriverAdapterBase | AsyncDriverAdapterBase

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

This method requires the connection to already exist in scope (e.g., from DI injection). For on-demand connection creation, use provide_request_session_sync or provide_request_session_async instead.

Parameters:
Returns:

A driver session instance for the specified database configuration.

provide_request_session_sync(key, state, scope)[source]
Overloads:

  • self, key (SyncDatabaseConfig[Any, Any, DriverT] | NoPoolSyncConfig[Any, DriverT]), state (State), scope (Scope) → DriverT

  • self, key (type[SyncDatabaseConfig[Any, Any, DriverT] | NoPoolSyncConfig[Any, DriverT]]), state (State), scope (Scope) → DriverT

  • self, key (str), state (State), scope (Scope) → SyncDriverAdapterBase

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

If no connection exists in scope, one will be created from the pool and stored in scope for reuse. The connection will be cleaned up by the before_send handler.

For async configurations, use provide_request_session_async instead.

Parameters:
Returns:

A sync driver session instance for the specified database configuration.

async provide_request_session_async(key, state, scope)[source]
Overloads:

  • self, key (AsyncDatabaseConfig[Any, Any, DriverT] | NoPoolAsyncConfig[Any, DriverT]), state (State), scope (Scope) → DriverT

  • self, key (type[AsyncDatabaseConfig[Any, Any, DriverT] | NoPoolAsyncConfig[Any, DriverT]]), state (State), scope (Scope) → DriverT

  • self, key (str), state (State), scope (Scope) → AsyncDriverAdapterBase

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

If no connection exists in scope, one will be created from the pool and stored in scope for reuse. The connection will be cleaned up by the before_send handler.

For sync configurations, use provide_request_session instead.

Parameters:
Returns:

An async driver session instance for the specified database configuration.

provide_request_connection(key, state, scope)[source]
Overloads:

  • self, key (SyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolSyncConfig[ConnectionT, Any] | AsyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolAsyncConfig[ConnectionT, Any]), state (State), scope (Scope) → ConnectionT

  • self, key (type[SyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolSyncConfig[ConnectionT, Any] | AsyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolAsyncConfig[ConnectionT, Any]]), state (State), scope (Scope) → ConnectionT

  • self, key (str), state (State), scope (Scope) → Any

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

This method requires the connection to already exist in scope (e.g., from DI injection). For on-demand connection creation, use provide_request_connection_sync or provide_request_connection_async instead.

Parameters:
Returns:

A database connection instance for the specified database configuration.

provide_request_connection_sync(key, state, scope)[source]
Overloads:

  • self, key (SyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolSyncConfig[ConnectionT, Any]), state (State), scope (Scope) → ConnectionT

  • self, key (type[SyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolSyncConfig[ConnectionT, Any]]), state (State), scope (Scope) → ConnectionT

  • self, key (str), state (State), scope (Scope) → Any

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

If no connection exists in scope, one will be created from the pool and stored in scope for reuse. The connection will be cleaned up by the before_send handler.

For async configurations, use provide_request_connection_async instead.

Parameters:
Returns:

A database connection instance for the specified database configuration.

async provide_request_connection_async(key, state, scope)[source]
Overloads:

  • self, key (AsyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolAsyncConfig[ConnectionT, Any]), state (State), scope (Scope) → ConnectionT

  • self, key (type[AsyncDatabaseConfig[ConnectionT, Any, Any] | NoPoolAsyncConfig[ConnectionT, Any]]), state (State), scope (Scope) → ConnectionT

  • self, key (str), state (State), scope (Scope) → Any

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

If no connection exists in scope, one will be created from the pool and stored in scope for reuse. The connection will be cleaned up by the before_send handler.

For sync configurations, use provide_request_connection instead.

Parameters:
Returns:

A database connection instance for the specified database configuration.

Configuration

class sqlspec.extensions.litestar.LitestarConfig[source]

Bases: TypedDict

Configuration options for Litestar session store extension.

All fields are optional with sensible defaults. Use in extension_config["litestar"]:

Example

from sqlspec.adapters.oracledb import OracleAsyncConfig

config = OracleAsyncConfig(

connection_config={"dsn": "oracle://localhost/XEPDB1"}, extension_config={

"litestar": {

"session_table": "my_sessions", "in_memory": True

}

}

)

Notes

This TypedDict provides type safety for extension config but is not required. You can use plain dicts as well.

session_table: NotRequired[str]

'litestar_session'

Examples

"app_sessions" "user_sessions" "tenant_acme_sessions"

Type:

Name of the sessions table. Default

in_memory: NotRequired[bool]

False.

When enabled, tables are created with the in-memory attribute for databases that support it.

This is an Oracle-specific feature that requires:

  • Oracle Database 12.1.0.2 or higher

  • Database In-Memory option license (Enterprise Edition)

  • Sufficient INMEMORY_SIZE configured in the database instance

Other database adapters ignore this setting.

Examples

Oracle with in-memory enabled:
config = OracleAsyncConfig(

connection_config={"dsn": "oracle://..."}, extension_config={

"litestar": {

"in_memory": True

}

}

)

Notes

  • Tables created with INMEMORY PRIORITY HIGH clause

  • Ignored by unsupported adapters

Type:

Enable in-memory table storage (Oracle-specific). Default

shard_count: NotRequired[int]

Optional hash shard count for session table primary key.

When set (>1), adapters that support computed shard columns (e.g., Spanner) will create a generated shard_id using MOD(FARM_FINGERPRINT(session_id), shard_count) and include it in the primary key to reduce hotspotting. Ignored by adapters that do not support computed shards.

table_options: NotRequired[str]

Optional raw OPTIONS/engine-specific table options string.

Passed verbatim when the adapter supports table-level OPTIONS/clauses (e.g., Spanner columnar/tiered storage). Ignored by adapters that do not support table options.

index_options: NotRequired[str]

Optional raw OPTIONS/engine-specific options for the expires_at index.

Passed verbatim to the index definition for adapters that support index OPTIONS/clauses. Ignored by adapters that do not support index options.

Channels Backend

class sqlspec.extensions.litestar.SQLSpecChannelsBackend[source]

Bases: ChannelsBackend

A Litestar Channels backend implemented on top of SQLSpec's EventChannel.

This backend allows Litestar's ChannelsPlugin to use a SQLSpec database as the broker. Under the hood it relies on SQLSpec's events extension, which can be configured to use a durable table queue or native adapter backends.

Notes

Litestar channels may use arbitrary string names. SQLSpec event channel names must be valid identifiers. This backend maps Litestar channel names to deterministic database channel identifiers via hashing.

__init__(event_channel, *, channel_prefix='litestar', poll_interval=0.2)[source]
async on_startup()[source]

Called by the plugin on application startup

Return type:

None

async on_shutdown()[source]

Called by the plugin on application shutdown

Return type:

None

async publish(data, channels)[source]

Publish the message data to all channels

Return type:

None

async subscribe(channels)[source]

Start listening for events on channels

Return type:

None

async unsubscribe(channels)[source]

Stop listening for events on channels

Return type:

None

stream_events()[source]

Return a generator, iterating over events of subscribed channels as they become available

Return type:

AsyncGenerator[tuple[str, bytes], None]

async get_history(channel, limit=None)[source]

Return history entries for a channel.

SQLSpec's event queue is primarily designed for durable delivery, not for history replay. For now, this backend does not expose history.

Return type:

list[bytes]

Store

class sqlspec.extensions.litestar.BaseSQLSpecStore[source]

Bases: ABC, Generic[ConfigT]

Base class for SQLSpec-backed Litestar session stores.

Implements the litestar.stores.base.Store protocol for server-side session storage using SQLSpec database adapters.

This abstract base class provides common functionality for all database-specific store implementations including: - Connection management via SQLSpec configs - Session expiration calculation - Table creation utilities

Subclasses must implement dialect-specific SQL queries.

Parameters:

config (TypeVar(ConfigT)) -- SQLSpec database configuration with extension_config["litestar"] settings.

Example

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

config = AsyncpgConfig(

connection_config={"dsn": "postgresql://..."}, extension_config={"litestar": {"session_table": "my_sessions"}}

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

Notes

Configuration is read from config.extension_config["litestar"]: - session_table: Table name (default: "litestar_session")

__init__(config)[source]

Initialize the session store.

Parameters:

config (TypeVar(ConfigT)) -- SQLSpec database configuration.

Notes

Reads table_name from config.extension_config["litestar"]["session_table"]. Defaults to "litestar_session" if not specified.

property config: ConfigT

Return the database configuration.

property table_name: str

Return the session table name.

abstractmethod 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 and the value had an initial expiry time set, renew the expiry time for renew_for seconds. If the value has not been set with an expiry time this is a no-op.

Return type:

bytes | None

Returns:

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

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

Store a session value.

Parameters:

  • key (str) -- Session ID.

  • value (str | bytes) -- Session data (will be converted to bytes if string).

  • expires_in (int | timedelta | None) -- Time in seconds or timedelta before expiration.

Return type:

None

abstractmethod async delete(key)[source]

Delete a session by key.

Parameters:

key (str) -- Session ID to delete.

Return type:

None

abstractmethod async delete_all()[source]

Delete all sessions from the store.

Return type:

None

abstractmethod 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.

abstractmethod 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.

abstractmethod async delete_expired()[source]

Delete all expired sessions.

Return type:

int

Returns:

Number of sessions deleted.

abstractmethod async create_table()[source]

Create the session table if it doesn't exist.

Return type:

None

async __aenter__()[source]

Enter context manager.

Return type:

Self

async __aexit__(exc_type, exc_val, exc_tb)[source]

Exit context manager.

Return type:

None

Providers

create_filter_dependencies() accepts camel-case aliases for configured orderBy fields by default. Use sort_field_aliases to map explicit API names to configured SQL-facing fields, or set sort_field_camelize=False when an endpoint must accept only raw configured values. Alias values are normalized before OrderByFilter is created, and unknown aliases cannot bypass the sort_field allowlist.

class sqlspec.extensions.litestar.providers.DependencyDefaults[source]

Bases: object

class sqlspec.extensions.litestar.providers.FilterConfig[source]

Bases: TypedDict

Configuration for generated Litestar filter dependencies.

All keys are optional. A filter dependency is created only for each enabled key. Field names are SQL-facing allowlist values; generated query parameter names and order-by aliases remain API-facing.

id_filter: NotRequired[type[UUID | int | str]]

Type of ID filter to enable. When set, creates an ids collection filter.

id_field: NotRequired[str]

SQL-facing field name for ID filtering. Defaults to "id".

sort_field: NotRequired[str | set[str] | list[str]]

Allowed SQL-facing field or fields for orderBy sorting.

sort_field_aliases: NotRequired[dict[str, str]]

Additional API-facing orderBy aliases mapped to configured sort_field values.

sort_field_camelize: NotRequired[bool]

Whether to accept camel-case aliases for configured sort fields. Defaults to True.

sort_order: NotRequired[Literal['asc', 'desc']]

Default sort order. Defaults to "desc".

pagination_type: NotRequired[Literal['limit_offset']]

Pagination strategy to enable. Currently supports "limit_offset".

pagination_size: NotRequired[int]

Default page size for limit/offset pagination.

search: NotRequired[str | set[str] | list[str]]

SQL-facing field or fields to search. Strings may be comma-separated.

search_ignore_case: NotRequired[bool]

Whether search filtering is case-insensitive. Defaults to False.

created_at: NotRequired[bool]

Whether to enable created_at before/after range filtering.

updated_at: NotRequired[bool]

Whether to enable updated_at before/after range filtering.

not_in_fields: NotRequired[FieldNameType | set[FieldNameType] | list[str | FieldNameType]]

Field or fields that support NOT IN collection filtering.

in_fields: NotRequired[FieldNameType | set[FieldNameType] | list[str | FieldNameType]]

Field or fields that support IN collection filtering.

null_fields: NotRequired[str | set[str] | list[str]]

Field or fields that support IS NULL filtering.

not_null_fields: NotRequired[str | set[str] | list[str]]

Field or fields that support IS NOT NULL filtering.

CLI

sqlspec.extensions.litestar.database_group

Click command group for managing SQLSpec database components (migrations, etc.).

Type:

click.Group