Extensions¶

SQLSpec extensions integrate the registry with frameworks and services such as Litestar and Google ADK. Use these extensions for dependency injection, lifecycle hooks, and application-facing helpers.

Overview¶

Google ADK: Session, event, and memory storage.
Litestar: Plugin-based integration with dependency injection and lifecycle management.
FastAPI/Flask/Starlette/Sanic: Framework helpers (see usage docs).

ADK Example¶

adk memory store¶

    from sqlspec.adapters.aiosqlite import AiosqliteConfig
    from sqlspec.adapters.aiosqlite.adk import AiosqliteADKStore

    config = AiosqliteConfig(connection_config={"database": ":memory:"})
    store = AiosqliteADKStore(config)
    await store.ensure_tables()

    session = await store.create_session(
        session_id="session_1", app_name="docs", user_id="user_1", state={"mode": "demo"}
    )

    event_record = {
        "id": "evt_1",
        "app_name": "docs",
        "user_id": "user_1",
        "session_id": session["id"],
        "invocation_id": "inv_1",
        "author": "user",
        "branch": "main",
        "actions": b"",
        "long_running_tool_ids_json": None,
        "timestamp": datetime.now(timezone.utc),
        "content": {"text": "Hello"},
        "grounding_metadata": None,
        "custom_metadata": None,
        "partial": False,
        "turn_complete": True,
        "interrupted": None,
        "error_code": None,
        "error_message": None,
    }
    await store.append_event(event_record)
    events = await store.get_events(session_id=session["id"])

Litestar Example¶

litestar plugin¶

from litestar import Litestar, get

from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig, SqliteDriver
from sqlspec.extensions.litestar import SQLSpecPlugin

sqlspec = SQLSpec()
sqlspec.add_config(
    SqliteConfig(
        connection_config={"database": ":memory:"}, extension_config={"litestar": {"session_key": "db_session"}}
    )
)

@get("/health")
def health_check(db_session: SqliteDriver) -> dict[str, str]:
    result = db_session.execute("SELECT 'ok' as status")
    return result.one()

app = Litestar(route_handlers=[health_check], plugins=[SQLSpecPlugin(sqlspec=sqlspec)])

Reference Modules¶

class sqlspec.extensions.adk.SQLSpecSessionService[source]

Bases: BaseSessionService

SQLSpec-backed implementation of BaseSessionService.

Provides session and event storage using SQLSpec database adapters. Delegates all database operations to a store implementation.

Parameters:: store¶ (BaseAsyncADKStore) – Database store implementation (e.g., AsyncpgADKStore).

Example

from sqlspec.adapters.asyncpg import AsyncpgConfig from sqlspec.adapters.asyncpg.adk.store import AsyncpgADKStore from sqlspec.extensions.adk.service import SQLSpecSessionService

config = AsyncpgConfig(connection_config={“dsn”: “postgresql://…”}) store = AsyncpgADKStore(config) await store.ensure_tables()

service = SQLSpecSessionService(store) session = await service.create_session(

app_name=”my_app”, user_id=”user123”, state={“key”: “value”}

)

__init__(store)[source]

Initialize the session service.

Parameters:: store¶ (BaseAsyncADKStore) – Database store implementation.

property store: BaseAsyncADKStore: Return the database store.

async create_session(*, app_name, user_id, state=None, session_id=None)[source]

Create a new session.

Parameters:

app_name¶ (str) – Name of the application.
user_id¶ (str) – ID of the user.
state¶ (dict[str, typing.Any] | None) – Initial state of the session.
session_id¶ (str | None) – Client-provided session ID. If None, generates a UUID.

Return type:

Session

Returns:

The newly created session.

async get_session(*, app_name, user_id, session_id, config=None)[source]

Get a session by ID.

Parameters:

app_name¶ (str) – Name of the application.
user_id¶ (str) – ID of the user.
session_id¶ (str) – Session identifier.
config¶ (GetSessionConfig | None) – Configuration for retrieving events.

Return type:

Session | None

Returns:

Session object if found, None otherwise.

async list_sessions(*, app_name, user_id=None)[source]

List all sessions for an app, optionally filtered by user.

Parameters:

app_name¶ (str) – Name of the application.
user_id¶ (str | None) – ID of the user. If None, all sessions for the app are listed.

Return type:

ListSessionsResponse

Returns:

Response containing list of sessions (without events).

async delete_session(*, app_name, user_id, session_id)[source]

Delete a session and all its events.

Parameters:

app_name¶ (str) – Name of the application.
user_id¶ (str) – ID of the user.
session_id¶ (str) – Session identifier.

Return type:

None

async append_event(session, event)[source]

Append an event to a session.

Parameters:

session¶ (Session) – Session to append to.
event¶ (Event) – Event to append.

Return type:

Event

Returns:

The appended event.

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]

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.

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:

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 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_request_session_sync(key, state, scope)[source]

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:

key¶ – The configuration identifier (same as get_config).
state¶ – The Litestar application State object.
scope¶ – 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.

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¶ – The configuration identifier (same as get_config).
state¶ – The Litestar application State object.
scope¶ – 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.

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:

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

Any

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.

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.

Return type:

Any

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.

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.

Return type:

Any

Returns:

A database connection instance for the specified database configuration.

Extensions¶

Overview¶

ADK Example¶

Litestar Example¶

Reference Modules¶

See Also¶