API#

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.

__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[Any, Any, Any] | NoPoolSyncConfig[Any, Any] | AsyncDatabaseConfig[Any, Any, Any] | NoPoolAsyncConfig[Any, 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[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any]]]]
Returns:: List of annotations.

get_annotation(key)[source]

Return the annotation for the given configuration.

Parameters:: key¶ (Union[str, SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any], type[Union[SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any]]]]) -- The configuration instance or key to lookup.
Raises:: KeyError -- If no configuration is found for the given key.
Return type:: type[Union[SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any]]]
Returns:: The annotation for the configuration.

get_config(name)[source]

Get a configuration instance by name.

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 (SyncConfigT) → SyncConfigT
self, name (AsyncConfigT) → AsyncConfigT
self, name (str) → AnyDatabaseConfig

Registry identifiers are available as soon as the plugin is constructed: a config instance, its concrete config type when exactly one configuration has that type, or a non-null bind_key. A bind_key wins over a dependency key of the same value. The generated Litestar dependency keys (session_key, connection_key, pool_key) resolve only after the plugin is registered with a Litestar application.

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, or if several configurations share the requested concrete type.
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.

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

This method requires the connection to already exist in scope. For on-demand connection creation, use provide_request_session_sync or provide_request_session_async instead.

Parameters:

key¶ (Union[str, SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any], type[Union[SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], 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.

Returns:

A driver session instance for the specified database configuration.

provide_request_session_sync(key, state, scope)[source]

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

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

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:

key¶ (Union[str, SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], type[Union[SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.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.

Returns:

A sync driver session instance for the specified database configuration.

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

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

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

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:

key¶ (Union[str, AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any], type[Union[AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.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.

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.

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

This method requires the connection to already exist in scope. For on-demand connection creation, use provide_request_connection_sync or provide_request_connection_async instead.

Parameters:

key¶ (Union[str, SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], AsyncDatabaseConfig[Any, Any, Any], NoPoolAsyncConfig[Any, Any], type[Union[SyncDatabaseConfig[Any, Any, Any], NoPoolSyncConfig[Any, Any], 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.

Returns:

A database connection instance for the specified database configuration.

provide_request_connection_sync(key, state, scope)[source]

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

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

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:

key¶ (Union[str, SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.Any], type[Union[SyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolSyncConfig[typing.Any, typing.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.

Returns:

A database connection instance for the specified database configuration.

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

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

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

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:

key¶ (Union[str, AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.Any], type[Union[AsyncDatabaseConfig[typing.Any, typing.Any, typing.Any], NoPoolAsyncConfig[typing.Any, typing.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.

Returns:

A database connection instance for the specified database configuration.

Session Stores#

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.

extension_config_options: ClassVar[frozenset[str]] = frozenset({'auto_trace_headers', 'commit_mode', 'connection_key', 'correlation_header', 'correlation_headers', 'create_schema', 'disable_di', 'enable_correlation_middleware', 'enable_sqlcommenter_middleware', 'extra_commit_statuses', 'extra_rollback_statuses', 'manage_schema', 'migrations_path', 'pool_key', 'pragma_overrides', 'pragma_profile', 'run_migrations', 'session_key', 'session_table'})

__init__(config)[source]

Initialize AioSQLite session store.

Parameters:: config¶ (AiosqliteConfig) -- AiosqliteConfig instance.

async create_table()[source]

Create the session table if it doesn't exist.

Return type:: None

async prepare_schema_async(driver)[source]

Apply configured SQLite PRAGMAs before migration DDL generation.

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:

key¶ (str) -- Session ID.
value¶ (str | bytes) -- Session data.
expires_in¶ (int | timedelta | None) -- Time until expiration.

Return type:

None

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.