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.

__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:
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:
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:
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:
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:
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:
Returns:

A database connection instance for the specified database configuration.

Configuration#

class sqlspec.extensions.litestar.LitestarConfig[source]#

Bases: TypedDict

Configuration options for Litestar SQLSpec plugin.

All fields are optional with sensible defaults.

migrations_path: NotRequired[str | Path]#

Directory containing this extension's migrations, or a '<dotted.module>:<subdir>' specification.

Overrides the default sqlspec.extensions.<name> lookup. Setting this auto-includes the extension in migration_config["include_extensions"].

session_table: NotRequired[bool | str]#

Enable session table for server-side session storage.

  • True: Use default table name ('litestar_session')

  • "custom_name": Use custom table name

When set, litestar extension migrations are auto-included to create the session table. If you're only using litestar for DI/connection management (not session storage), leave this unset to skip the migrations.

connection_key: NotRequired[str]#

'db_connection'

Type:

Key for storing connection in ASGI scope. Default

pool_key: NotRequired[str]#

'db_pool'

Type:

Key for storing connection pool in application state. Default

session_key: NotRequired[str]#

'db_session'

Type:

Key for storing session in ASGI scope. Default

commit_mode: NotRequired[Literal['manual', 'autocommit', 'autocommit_include_redirect']]#

'manual'

Type:

Transaction commit mode. Default

enable_correlation_middleware: NotRequired[bool]#

True

Type:

Enable request correlation ID middleware. Default

correlation_header: NotRequired[str]#

X-Request-ID

Type:

HTTP header to read the request correlation ID from when middleware is enabled. Default

correlation_headers: NotRequired[tuple[str, ...] | list[str]]#

Additional HTTP headers to read as correlation ID fallbacks.

auto_trace_headers: NotRequired[bool]#

True.

Type:

Read standard trace context headers as correlation ID fallbacks. Default

extra_commit_statuses: NotRequired[set[int]]#
Type:

Additional HTTP status codes that trigger commit. Default

extra_rollback_statuses: NotRequired[set[int]]#
Type:

Additional HTTP status codes that trigger rollback. Default

disable_di: NotRequired[bool]#

False. When True, the Litestar plugin will not register dependency providers for database connections, pools, and sessions, or the per-request handler that commits and closes request connections. Pool startup and shutdown follow manage_lifespan, which defaults to False when disable_di is True.

Type:

Disable built-in dependency injection. Default

manage_lifespan: NotRequired[bool]#

the inverse of disable_di. When True, the Litestar plugin creates the configuration's pool on application startup, stores it in application state under pool_key, and closes it on shutdown, whether or not disable_di is set. Set to True alongside disable_di=True to keep pool lifecycle management while another DI solution provides connections and sessions. When False, the application creates and closes the pool itself; with dependency injection enabled, the plugin's providers still read the pool from application state under pool_key.

Type:

Register the plugin's pool lifespan handler. Default

enable_sqlcommenter_middleware: NotRequired[bool]#

True. When the driver's StatementConfig has enable_sqlcommenter=True, the middleware is registered automatically. Set to False to explicitly disable middleware registration even when SQLCommenter is enabled on the driver config.

Type:

Control automatic SQLCommenter middleware registration. Default

manage_schema: NotRequired[bool]#

True.

Type:

Apply additive session-table reconciliation. Default

create_schema: NotRequired[bool]#

True.

Type:

Create a missing session table during managed reconciliation. Default

Correlation Middleware#

SQLSpecPlugin installs CorrelationMiddleware while enable_correlation_middleware is enabled, which is the default. Add it yourself only when you build the middleware stack without the plugin.

When the application's middleware stack already contains CorrelationMiddleware or a subclass, SQLSpecPlugin automatically skips adding a duplicate instance. It logs at DEBUG when the existing middleware honors the configured headers, or logs a WARNING when configured correlation headers would remain unapplied.

from litestar.middleware import DefineMiddleware

from sqlspec.extensions.litestar import TRACE_CONTEXT_FALLBACK_HEADERS, CorrelationMiddleware

middleware = [DefineMiddleware(CorrelationMiddleware, headers=TRACE_CONTEXT_FALLBACK_HEADERS)]
class sqlspec.extensions.litestar.CorrelationMiddleware[source]#

Bases: object

ASGI middleware that binds a correlation ID to each HTTP request.

The first non-empty value among headers, checked in order, becomes the correlation ID after surrounding whitespace is trimmed and the value is truncated to 128 characters; when no header has a value, or the trimmed value is empty, a new ID is generated. The ID is set on CorrelationContext and in the request scope for the duration of the request, then the previous ID is restored. Non-HTTP scopes, and instances created with an empty headers tuple, pass through unchanged.

Parameters:
__init__(app, *, headers)[source]#
sqlspec.extensions.litestar.TRACE_CONTEXT_FALLBACK_HEADERS#

Ordered trace-context header names that the plugin checks after correlation_header and correlation_headers while auto_trace_headers is enabled, which is the default. Import the constant to inspect the names or pass them to CorrelationMiddleware.

Type:

tuple[str, ...]

Error Handling#

SQLSpecPlugin automatically registers default exception handlers on Litestar's application configuration for database errors:

  • NotFoundError: translated to HTTP 404, with the error message as detail.

  • IntegrityError (and subclasses such as UniqueViolationError): translated to HTTP 409 Conflict with the generic detail "Conflict", preventing internal database constraint or schema text from leaking to clients.

The plugin renders these responses within the route's middleware stack, keeping headers added by application middleware and running after_exception hooks once. Handlers explicitly registered on the application or router for IntegrityError or its subclasses take precedence; handlers for broader exceptions (such as SQLSpecError, Exception, or status 500) receive the original exception, while handlers for status 409 or litestar.exceptions.HTTPException render the HTTP response.

sqlspec.extensions.litestar.plugin.not_found_error_handler(request, exc)[source]#

Return a 404 response whose detail is the NotFoundError message.

A handler for status 404 or for the HTTP exception type renders it, otherwise Litestar's default exception response. Handlers for NotFoundError or a subclass take precedence; handlers for its base classes or status 500 do not.

Return type:

Response[typing.Any]

sqlspec.extensions.litestar.plugin.integrity_error_handler(request, exc)[source]#

Return a 409 response with the generic detail "Conflict".

A handler for status 409 or for the HTTP exception type renders it, otherwise Litestar's default exception response. Handlers for IntegrityError or a subclass take precedence, and handlers for a later class in the error's MRO or for status 500 receive the original exception.

Return type:

Response[typing.Any]

Channels Backend#

SQLSpecChannelsBackend buffers decoded output from every asynchronous EventChannel transport. Pass output_queue_capacity to bound that buffer; the default None remains unbounded. When full, the backend discards the oldest decoded message before acknowledging and retaining the newest one. output_queue_depth reports the current pending count and dropped_message_count reports cumulative overflow drops for the backend instance. Malformed payloads are acknowledged and logged without increasing the overflow count. Shutdown clears pending output while preserving the cumulative drop diagnostic for lifecycle reuse.

metrics_snapshot() returns every observability metric recorded for the event channel's database configuration, including loader, migration, storage, and other event channels on that configuration, merged with channels.output_queue_depth and channels.dropped_message_count. Those two keys are unprefixed and describe this backend instance only; they mirror the output_queue_depth and dropped_message_count properties. All values are floats.

Payload budget#

Each published payload is base64-wrapped as {"data_b64": ...}. The PostgreSQL notify event backend sends that payload inside a notification envelope and rejects an envelope larger than MAX_NOTIFY_BYTES. measure(data) returns the encoded size of the notify envelope that wraps data, and fits(data) reports whether it is within notify_budget. notify_budget is None for every other backend kind (notify_queue, poll_queue, aq, txeventq), so fits() always returns True there. Check a payload before publishing and send oversized data as smaller messages or as a compact reference the subscriber resolves:

backend = SQLSpecChannelsBackend(spec.event_channel(config))

if backend.fits(payload):
    await backend.publish(payload, ["updates"])
else:
    await backend.publish(b'{"object_key": "updates/123.json"}', ["updates"])

API#

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.

__init__(event_channel, *, channel_prefix='litestar', poll_interval=0.2, output_queue_capacity=None)[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

property output_queue_depth: int#

Return the number of buffered channel messages.

property dropped_message_count: int#

Return the cumulative number of messages dropped due to overflow.

property notify_budget: int | None#

Return the NOTIFY byte budget, or None when the event backend has no payload limit.

Only the notify backend carries each payload inside a PostgreSQL notification; every other backend kind returns None.

measure(data)[source]#

Return the encoded size of the PostgreSQL notify envelope that wraps data.

Parameters:

data (bytes) -- Channel payload to measure.

Return type:

int

Returns:

The UTF-8 byte size of the notify envelope carrying the base64-wrapped data.

fits(data)[source]#

Return whether data fits the event backend's payload budget.

Parameters:

data (bytes) -- Channel payload to check.

Return type:

bool

Returns:

True when the backend has no payload budget or the encoded envelope is within it.

metrics_snapshot()[source]#

Return configuration metrics merged with this backend instance's queue counters.

Return type:

dict[str, float]

Returns:

Every observability metric recorded for the event channel's database configuration, plus this backend's channels.output_queue_depth and channels.dropped_message_count.

async publish(data, channels)[source]#

Publish the message data to all channels

Return type:

None

async publish_many(data, channels)[source]#

Publish independent payloads through one event-channel batch.

Parameters:
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: Store, Generic[ConfigT]

Base class for SQLSpec-backed Litestar session stores.

Inherits Litestar's Store abstract base class for server-side session storage backed by 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.

__init__(config)[source]#

Initialize the session store.

Parameters:

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

property config: ConfigT#

Return the database configuration.

property table_name: str#

Return the session table name.

property create_schema_enabled: bool#

Return whether adapter-level creation should run.

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

prepare_schema_sync(driver)[source]#

Prepare adapter-specific schema decisions with a synchronous driver.

Return type:

None

async prepare_schema_async(driver)[source]#

Prepare adapter-specific schema decisions with an asynchronous driver.

Return type:

None

async reconcile_schema(*, assume_existing=False)[source]#

Apply additive session-table changes from the canonical store DDL.

Parameters:

assume_existing (bool) -- Skip table discovery after adapter-level creation.

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.

sqlspec.extensions.litestar.providers.create_filter_dependencies(config, dep_defaults=<sqlspec.extensions.litestar.providers.DependencyDefaults object>)[source]#

Create a dependency provider for the combined filter function.

Parameters:
Return type:

dict[str, Provide]

Returns:

A dependency provider function for the combined filter function.

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', 'cursor']]#

"limit_offset" or "cursor".

Type:

Pagination strategy to enable

cursor_keys: NotRequired[str | Sequence[CursorKey | str | tuple[str, Literal['asc', 'desc']]]]#

Field names, (field, direction) pairs, or CursorKey values. The last key must be unique per row.

cursor_secret: NotRequired[str | bytes]#

Secret enabling HMAC-SHA256 signed cursors.

pagination_size: NotRequired[int]#

Default page size for limit/offset or cursor pagination.

pagination_max_size: NotRequired[int]#

Largest page size a client may request. Defaults to 1000.

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.

boolean_fields: NotRequired[str | set[str] | list[str]]#

Field or fields that support boolean filtering.

choice_fields: NotRequired[ChoiceField | set[ChoiceField] | list[str | ChoiceField]]#

Field or fields that support choices filtering.

CLI#

sqlspec.extensions.litestar.database_group#

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

Type:

click.Group