Extensions ¶

abstractmethod 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, returns all sessions for the app.

Return type:

list[SessionRecord]

Returns:

List of session records.

abstractmethod async delete_session(session_id)[source]

Delete a session and its events.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: None

abstractmethod async append_event(event_record)[source]

Append an event to a session.

Parameters:: event_record¶ (EventRecord) – Event record to store.
Return type:: None

abstractmethod async get_events(session_id, after_timestamp=None, limit=None)[source]

Get events for a session.

Parameters:

session_id¶ (str) – Session identifier.
after_timestamp¶ (datetime | None) – Only return events after this time.
limit¶ (int | None) – Maximum number of events to return.

Return type:

list[EventRecord]

Returns:

List of event records ordered by timestamp ascending.

abstractmethod async create_tables()[source]

Create the sessions and events tables if they don’t exist.

Return type:: None

class sqlspec.extensions.adk.BaseSyncADKStore[source]

Bases: ABC, Generic[ConfigT]

Base class for sync SQLSpec-backed ADK session stores.

Implements storage operations for Google ADK sessions and events using SQLSpec database adapters with synchronous execution.

This abstract base class provides common functionality for sync database-specific store implementations including: - Connection management via SQLSpec configs - Table name validation - Session and event CRUD operations

Subclasses must implement dialect-specific SQL queries and will be created in each adapter directory (e.g., sqlspec/adapters/sqlite/adk/store.py).

Parameters:: config¶ (TypeVar(ConfigT)) – SQLSpec database configuration with extension_config[“adk”] settings.

Notes

Abstract base class for sync ADK session stores. See API Reference for details.

__init__(config)[source]

Initialize the sync ADK store.

Parameters:: config¶ (TypeVar(ConfigT)) – SQLSpec database configuration.

Notes

property config: ConfigT: Return the database configuration.

property session_table: str: Return the sessions table name.

property events_table: str: Return the events table name.

property owner_id_column_ddl: str | None: Return the full owner ID column DDL (or None if not configured).

property owner_id_column_name: str | None: Return the owner ID column name only (or None if not configured).

abstractmethod create_session(session_id, app_name, user_id, state, owner_id=None)[source]

Create a new session.

Parameters:

session_id¶ (str) – Unique identifier for the session.
app_name¶ (str) – Name of the application.
user_id¶ (str) – ID of the user.
state¶ (dict[str, typing.Any]) – Session state dictionary.
owner_id¶ (Optional[typing.Any]) – Optional owner ID value for owner_id_column (if configured).

Return type:

SessionRecord

Returns:

The created session record.

abstractmethod get_session(session_id)[source]

Get a session by ID.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: SessionRecord | None
Returns:: Session record if found, None otherwise.

abstractmethod update_session_state(session_id, state)[source]

Update session state.

Parameters:

session_id¶ (str) – Session identifier.
state¶ (dict[str, typing.Any]) – New state dictionary.

Return type:

abstractmethod 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, returns all sessions for the app.

Return type:

list[SessionRecord]

Returns:

List of session records.

abstractmethod delete_session(session_id)[source]

Delete a session and its events.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: None

abstractmethod create_event(event_id, session_id, app_name, user_id, author=None, actions=None, content=None, **kwargs)[source]

Create a new event.

Parameters:

event_id¶ (str) – Unique event identifier.
session_id¶ (str) – Session identifier.
app_name¶ (str) – Application name.
user_id¶ (str) – User identifier.
author¶ (str | None) – Event author (user/assistant/system).
actions¶ (bytes | None) – Pickled actions object.
content¶ (dict[str, typing.Any] | None) – Event content (JSONB/JSON).
**kwargs¶ (Any) – Additional optional fields.

Return type:

EventRecord

Returns:

Created event record.

abstractmethod list_events(session_id)[source]

List events for a session ordered by timestamp.

Parameters:: session_id¶ (str) – Session identifier.
Return type:: list[EventRecord]
Returns:: List of event records ordered by timestamp ASC.

abstractmethod create_tables()[source]

Create both sessions and events tables if they don’t exist.

Return type:: None

Session Service¶

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(pool_config={“dsn”: “postgresql://…”}) store = AsyncpgADKStore(config) await store.create_tables()

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

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

)

SQLSpec-backed implementation of Google ADK’s BaseSessionService. See API Reference for details.

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

SyncDriverAdapterBase | AsyncDriverAdapterBase

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.

Litestar Integration¶

The Litestar extension provides a plugin for SQLSpec integration with automatic dependency injection.

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(

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.

Main plugin for Litestar integration.

Features:

Automatic connection pool lifecycle
Dependency injection for drivers
Per-request session management
Transaction handling
Configuration from Litestar settings

Basic usage:

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

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

plugin = SQLSpecPlugin(sqlspec=sql)

@get("/users")
async def get_users(db: AsyncpgDriver) -> list[dict]:
    result = await db.select("SELECT * FROM users")
    return result.all()

app = Litestar(route_handlers=[get_users], plugins=[plugin])

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

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¶

Session Backend¶

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(: pool_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”)

Abstract base class for session storage backends.

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

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

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

Exit context manager.

Return type:: None

Starlette Integration¶

Starlette extension for SQLSpec.

Provides middleware-based session management, automatic transaction handling, and connection pooling lifecycle management for Starlette applications.

class sqlspec.extensions.starlette.SQLSpecAutocommitMiddleware[source]¶

Bases: BaseHTTPMiddleware

Middleware for autocommit transaction mode.

Acquires connection, commits on success status codes, rollbacks on error status codes.

__init__(app, config_state, include_redirect=False)[source]¶

Initialize middleware.

Parameters:

app¶ (Any) – Starlette application instance.
config_state¶ (SQLSpecConfigState) – Configuration state for this database.
include_redirect¶ (bool) – If True, commit on 3xx status codes as well.

async dispatch(request, call_next)[source]¶

Process request with autocommit transaction mode.

Parameters:

request¶ (Request) – Incoming HTTP request.
call_next¶ (Any) – Next middleware or route handler.

Return type:

Returns:

HTTP response.

class sqlspec.extensions.starlette.SQLSpecManualMiddleware[source]¶

Bases: BaseHTTPMiddleware

Middleware for manual transaction mode.

Acquires connection from pool, stores in request.state, releases after request. No automatic commit or rollback - user code must handle transactions.

__init__(app, config_state)[source]¶

Initialize middleware.

Parameters:

app¶ (Any) – Starlette application instance.
config_state¶ (SQLSpecConfigState) – Configuration state for this database.

async dispatch(request, call_next)[source]¶

Process request with manual transaction mode.

Parameters:

request¶ (Request) – Incoming HTTP request.
call_next¶ (Any) – Next middleware or route handler.

Return type:

Returns:

HTTP response.

class sqlspec.extensions.starlette.SQLSpecPlugin[source]¶

Bases: object

SQLSpec integration for Starlette applications.

Provides middleware-based session management, automatic transaction handling, and connection pooling lifecycle management.

Example

from starlette.applications import Starlette from sqlspec import SQLSpec from sqlspec.adapters.asyncpg import AsyncpgConfig from sqlspec.extensions.starlette import SQLSpecPlugin

sqlspec = SQLSpec() sqlspec.add_config(AsyncpgConfig(

bind_key=”default”, pool_config={“dsn”: “postgresql://localhost/mydb”}, extension_config={

“starlette”: {
“commit_mode”: “autocommit”, “session_key”: “db”

}

}

))

app = Starlette() db_ext = SQLSpecPlugin(sqlspec, app)

@app.route(“/users”) async def list_users(request):

db = db_ext.get_session(request) result = await db.execute(“SELECT * FROM users”) return JSONResponse({“users”: result.all()})

__init__(sqlspec, app=None)[source]¶

Initialize SQLSpec Starlette extension.

Parameters:

sqlspec¶ (SQLSpec) – Pre-configured SQLSpec instance with registered configs.
app¶ (Starlette | None) – Optional Starlette application to initialize immediately.

get_connection(request, key=None)[source]¶

Get database connection from request state.

Parameters:

request¶ (Request) – Starlette request instance.
key¶ (str | None) – Optional session key to retrieve specific database connection.

Return type:

Returns:

Database connection object.

get_session(request, key=None)[source]¶

Get or create database session for request.

Sessions are cached per request to ensure consistency.

Parameters:

request¶ (Request) – Starlette request instance.
key¶ (str | None) – Optional session key to retrieve specific database session.

Return type:

Returns:

Database session (driver instance).

init_app(app)[source]¶

Initialize Starlette application with SQLSpec.

Validates configuration, wraps lifespan, and adds middleware.

Parameters:: app¶ (Starlette) – Starlette application instance.
Return type:: None

lifespan(app)[source]¶

Manage connection pool lifecycle.

Parameters:: app¶ (Starlette) – Starlette application instance.
Yields:: None
Return type:: AsyncGenerator[None, None]

aiosql Integration¶

SQLSpec aiosql integration for loading SQL files.

This module provides a simple way to load aiosql-style SQL files and use them with SQLSpec drivers. It focuses on just the file parsing functionality, returning SQL objects that work with existing SQLSpec execution.

class sqlspec.extensions.aiosql.AiosqlAsyncAdapter[source]¶

Bases: _AiosqlAdapterBase[AsyncDriverAdapterBase], AsyncDriverAdapterProtocol

Asynchronous adapter that implements aiosql protocol using SQLSpec drivers.

This adapter bridges aiosql’s async driver protocol with SQLSpec’s async drivers, enabling queries loaded by aiosql to be executed with SQLSpec async drivers.

__init__(driver)[source]¶

Initialize the async adapter.

Parameters:: driver¶ (AsyncDriverAdapterBase) – SQLSpec async driver to use for execution

async insert_returning(conn, query_name, sql, parameters)[source]¶

Execute INSERT with RETURNING and return result.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type:

Returns:

Returned value or None

async insert_update_delete(conn, query_name, sql, parameters)[source]¶

Execute INSERT/UPDATE/DELETE.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type:

Note

Returns None per aiosql async protocol

async insert_update_delete_many(conn, query_name, sql, parameters)[source]¶

Execute INSERT/UPDATE/DELETE with many parameter sets.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Sequence of parameter sets

Return type:

Note

Returns None per aiosql async protocol

is_aio_driver: ClassVar[bool] = True¶

async select(conn, query_name, sql, parameters, record_class=None)[source]¶

Execute a SELECT query and return results as list.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters
record_class¶ (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

list[Any]

Returns:

List of query result rows

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

async select_cursor(conn, query_name, sql, parameters)[source]¶

Execute a SELECT query and return cursor context manager.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Yields:

Cursor-like object with results

Return type:

_AsyncCursorContextManager[typing.Any]

async select_one(conn, query_name, sql, parameters, record_class=None)[source]¶

Execute a SELECT query and return first result.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters
record_class¶ (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

Returns:

First result row or None

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

async select_value(conn, query_name, sql, parameters)[source]¶

Execute a SELECT query and return first value of first row.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type:

Returns:

First value of first row or None

class sqlspec.extensions.aiosql.AiosqlSyncAdapter[source]¶

Bases: _AiosqlAdapterBase[SyncDriverAdapterBase], SyncDriverAdapterProtocol

Synchronous adapter that implements aiosql protocol using SQLSpec drivers.

This adapter bridges aiosql’s synchronous driver protocol with SQLSpec’s sync drivers, enabling queries loaded by aiosql to be executed with SQLSpec drivers.

__init__(driver)[source]¶

Initialize the sync adapter.

Parameters:: driver¶ (SyncDriverAdapterBase) – SQLSpec sync driver to use for execution

insert_returning(conn, query_name, sql, parameters)[source]¶

Execute INSERT with RETURNING and return result.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type:

Returns:

Returned value or None

insert_update_delete(conn, query_name, sql, parameters)[source]¶

Execute INSERT/UPDATE/DELETE and return affected rows.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type:

int

Returns:

Number of affected rows

insert_update_delete_many(conn, query_name, sql, parameters)[source]¶

Execute INSERT/UPDATE/DELETE with many parameter sets.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Sequence of parameter sets

Return type:

int

Returns:

Number of affected rows

is_aio_driver: ClassVar[bool] = False¶

select(conn, query_name, sql, parameters, record_class=None)[source]¶

Execute a SELECT query and return results as generator.

Parameters:

conn¶ (Any) – Database connection (passed through to SQLSpec driver)
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters
record_class¶ (Any | None) – Deprecated - use schema_type in driver.execute instead

Yields:

Query result rows

Return type:

Generator[Any, None, None]

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

select_cursor(conn, query_name, sql, parameters)[source]¶

Execute a SELECT query and return cursor context manager.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Yields:

Cursor-like object with results

Return type:

Generator[Any, None, None]

select_one(conn, query_name, sql, parameters, record_class=None)[source]¶

Execute a SELECT query and return first result.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters
record_class¶ (Any | None) – Deprecated - use schema_type in driver.execute instead

Return type:

tuple[Any, ...] | None

Returns:

First result row or None

Note

The record_class parameter is ignored for compatibility. Use schema_type in driver.execute or _sqlspec_schema_type in parameters for type mapping.

select_value(conn, query_name, sql, parameters)[source]¶

Execute a SELECT query and return first value of first row.

Parameters:

conn¶ (Any) – Database connection
query_name¶ (str) – Name of the query
sql¶ (str) – SQL string
parameters¶ (Union[Dict[str, Any], List[Any], None]) – Query parameters

Return type: