aiosqlite#

Async SQLite adapter using aiosqlite. Supports URI-based in-memory databases with per-config instance isolation.

Configuration#

class sqlspec.adapters.aiosqlite.AiosqliteConfig[source]#

Bases: AsyncDatabaseConfig[AiosqliteConnection, AiosqliteConnectionPool, AiosqliteDriver]

Database configuration for AioSQLite engine.

driver_type#: alias of AiosqliteDriver

__init__(*, connection_config=None, connection_instance=None, migration_config=None, statement_config=None, driver_features=None, bind_key=None, extension_config=None, observability_config=None, **kwargs)[source]#

Initialize AioSQLite configuration.

Parameters:

connection_config¶ (AiosqlitePoolParams | dict[str, typing.Any] | None) -- Connection and pool configuration parameters (TypedDict or dict)
connection_instance¶ (AiosqliteConnectionPool | None) -- Optional pre-configured connection pool instance.
migration_config¶ (dict[str, typing.Any] | None) -- Optional migration configuration.
statement_config¶ (StatementConfig | None) -- Optional statement configuration.
driver_features¶ (AiosqliteDriverFeatures | dict[str, typing.Any] | None) -- Optional driver feature configuration.
bind_key¶ (str | None) -- Optional unique identifier for this configuration.
extension_config¶ (dict[str, dict[str, Any] | LitestarConfig | FastAPIConfig | StarletteConfig | SanicConfig | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) -- Extension-specific configuration
observability_config¶ (ObservabilityConfig | None) -- Adapter-level observability overrides for lifecycle hooks and observers
**kwargs¶ (Any) -- Additional keyword arguments passed to the base configuration.

get_signature_namespace()[source]#

Get the signature namespace for AiosqliteConfig types.

Return type:: dict[str, typing.Any]
Returns:: Dictionary mapping type names to types.

async create_connection()[source]#

Open a standalone connection owned by the caller.

The connection carries the same parameters, PRAGMAs, and runtime setup the pool applies, consumes no pool slot, and must be closed by the caller.

Return type:: Connection
Returns:: A newly opened aiosqlite connection.

async provide_pool()[source]#

Provide async pool instance.

Return type:: AiosqliteConnectionPool
Returns:: The async connection pool.

Connection Parameters#

class sqlspec.adapters.aiosqlite.AiosqliteConnectionParams[source]#

Bases: TypedDict

TypedDict for aiosqlite connection parameters.

Pool Parameters#

class sqlspec.adapters.aiosqlite.AiosqlitePoolParams[source]#

Bases: AiosqliteConnectionParams

TypedDict for aiosqlite pool parameters, inheriting connection parameters.

Driver Features#

class sqlspec.adapters.aiosqlite.AiosqliteDriverFeatures[source]#

Bases: TypedDict

Aiosqlite driver feature configuration.

Controls optional type handling and serialization features for SQLite connections.

enable_custom_adapters: Enable custom type adapters for JSON/UUID/datetime conversion.: Defaults to False. Set to True if you want global sqlite3 module type adapter/converter registration.
json_serializer: Custom JSON serializer function.: Defaults to sqlspec.utils.serializers.to_json.
json_deserializer: Custom JSON deserializer function.: Defaults to sqlspec.utils.serializers.from_json.
on_connection_create: Async callback executed when a connection is created.: Receives the raw aiosqlite connection for low-level driver configuration. Runs after internal setup (PRAGMA optimizations).
enable_events: Enable database event channel support.: Defaults to True when extension_config["events"] is configured. Provides pub/sub capabilities via table-backed queue (SQLite has no native pub/sub). Requires extension_config["events"] for migration setup.
events_backend: Event channel backend selection.: Only option: "poll_queue" (durable table-backed queue with lease-based retries and acknowledgements). SQLite does not have native pub/sub, so poll_queue is the only backend. Defaults to "poll_queue".
custom_functions: Register SQL functions that run on the aiosqlite worker thread.: Each entry must include name, narg, and func. Callable values are plain sync callables.
custom_collations: Register SQL collations that compare two string values.: Each entry must include name and func. Callable values are plain sync callables.
custom_aggregates: Register SQL aggregates with step/finalize classes.: Each entry must include name, narg, and aggregate_class.
custom_window_functions: Register user-defined aggregate window functions.: Each entry must include name, narg, and window_class.

default_transaction_mode: Default SQLite transaction mode (DEFERRED, IMMEDIATE, or EXCLUSIVE). authorizer_callback: sqlite3 authorizer hook run during statement compilation on the worker thread. trace_callback: sqlite3 trace hook run for executed statements on the worker thread. progress_handler: sqlite3 progress hook run every progress_handler_interval VM opcodes. progress_handler_interval: Progress callback interval in SQLite virtual machine opcodes.

Must be a positive integer when provided.

row_factory: Row factory selector or callable used for raw aiosqlite connections.: "row" maps to sqlite3.Row, "dict" maps to a dict row adapter, "tuple" keeps tuple rows. "dict" and custom callables can change raw connection result shapes seen by callers.

text_factory: Text factory used for raw aiosqlite connections. pragmas: Additional PRAGMA settings applied after built-in optimization PRAGMAs.

User values override built-in defaults when the same PRAGMA appears in both places.

extensions: Shared-library extension paths loaded on each connection.

User-Defined Functions and Extensions#

class sqlspec.adapters.aiosqlite.AiosqliteFunctionConfig[source]#

Bases: TypedDict

User-defined aiosqlite function registration.

class sqlspec.adapters.aiosqlite.AiosqliteCollationConfig[source]#

Bases: TypedDict

User-defined aiosqlite collation registration.

class sqlspec.adapters.aiosqlite.AiosqliteAggregateConfig[source]#

Bases: TypedDict

User-defined aiosqlite aggregate registration.

Driver#

class sqlspec.adapters.aiosqlite.AiosqliteDriver[source]#

Bases: AsyncDriverAdapterBase

AIOSQLite driver for async SQLite database operations.

__init__(connection, statement_config=None, driver_features=None)[source]#

Initialize driver adapter with connection and configuration.

Parameters:

connection¶ (Connection) -- Database connection instance
statement_config¶ (StatementConfig | None) -- Statement configuration for the driver
driver_features¶ (dict[str, typing.Any] | None) -- Driver-specific features like extensions, secrets, and connection callbacks
observability¶ -- Optional runtime handling lifecycle hooks, observers, and spans

property data_dictionary: AiosqliteDataDictionary#

Get the data dictionary for this driver.

Returns:: Data dictionary instance for metadata queries

async dispatch_execute(cursor, statement)[source]#

Execute single SQL statement.

Return type:: ExecutionResult

async dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets.

Return type:: ExecutionResult

async dispatch_execute_script(cursor, statement)[source]#

Execute SQL script.

Return type:: ExecutionResult

async execute_many(statement, /, parameters, *filters, statement_config=None, **kwargs)[source]#: Execute many with an AIOSQLite thin path for simple qmark batches.

async begin(mode=None)[source]#

Begin a database transaction.

Parameters:: mode¶ (Optional[Literal['DEFERRED', 'IMMEDIATE', 'EXCLUSIVE']]) -- Transaction locking mode (DEFERRED, IMMEDIATE, EXCLUSIVE). Falls back to driver_features['default_transaction_mode'], or IMMEDIATE when neither is set.
Return type:: None

async commit()[source]#

Commit the current transaction.

Return type:: None

async rollback()[source]#

Rollback the current transaction.

Return type:: None

with_cursor(connection)[source]#

Create async context manager for AIOSQLite cursor.

Return type:: AiosqliteCursor

dispatch_select_stream(statement, chunk_size)[source]#

Return a native aiosqlite row stream backed by chunked fetchmany.

Return type:: Optional[AsyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()[source]#

Handle AIOSQLite-specific exceptions.

Return type:: AiosqliteExceptionHandler

async select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)[source]#

Execute a query and stream Arrow results into storage.

Return type:: StorageBridgeJob

async load_from_arrow(table, source, *, batch_size=10000, partitioner=None, overwrite=False, telemetry=None)[source]#

Load Arrow data into SQLite using batched inserts.

Return type:: StorageBridgeJob

async load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)[source]#

Load staged artifacts from storage into SQLite.

Return type:: StorageBridgeJob

collect_rows(cursor, fetched)[source]#

Collect aiosqlite rows for the direct execution path.

Return type:: tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)[source]#

Resolve rowcount from aiosqlite cursor for the direct execution path.

Return type:: int

Connection Pool#

class sqlspec.adapters.aiosqlite.AiosqliteConnectionPool[source]#

Bases: object

Multi-connection pool for aiosqlite.

__init__(connection_parameters, pool_size=5, min_size=0, connect_timeout=30.0, idle_timeout=86400, operation_timeout=10.0, health_check_interval=30.0, enable_optimizations=True, enable_foreign_keys=False, on_connection_create=None, runtime_setup=None)[source]#

Initialize connection pool.

Parameters:

connection_parameters¶ -- SQLite connection parameters
pool_size¶ -- Maximum number of connections in the pool
min_size¶ -- Minimum connections to pre-create (pool warming)
connect_timeout¶ -- Maximum time to wait for connection acquisition
idle_timeout¶ -- Maximum time a connection can remain idle
operation_timeout¶ -- Maximum time for connection operations
health_check_interval¶ -- Seconds of idle time before running health check
enable_optimizations¶ -- Whether to apply performance PRAGMAs
enable_foreign_keys¶ -- Whether to enable foreign-key enforcement
on_connection_create¶ -- Async callback executed when connection is created
runtime_setup¶ -- Runtime feature setup to apply to new connections

property is_closed: bool#

Check if pool is closed.

Returns:: True if pool is closed

size()[source]#

Get total number of connections in pool.

Return type:: int
Returns:: Total connection count

checked_out()[source]#

Get number of checked out connections.

Return type:: int
Returns:: Number of connections currently in use

async new_connection()[source]#

Open a standalone connection configured exactly like a pooled one.

The result is owned by the caller: the pool neither tracks nor closes it.

Return type:: Connection
Returns:: A newly opened, fully configured connection.

async acquire()[source]#

Acquire a connection from the pool.

Return type:: AiosqlitePoolConnection
Returns:: Available connection
Raises:: AiosqliteConnectTimeoutError -- If acquisition times out

async release(connection)[source]#

Release a connection back to the pool.

Parameters:: connection¶ (AiosqlitePoolConnection) -- Connection to release
Return type:: None

get_connection()[source]#

Get a connection with automatic release.

Return type:: AiosqlitePoolConnectionContext

async close()[source]#

Close the connection pool.

Return type:: None

Data Dictionary#

class sqlspec.adapters.aiosqlite.data_dictionary.AiosqliteDataDictionary[source]#

Bases: AsyncDataDictionaryBase

SQLite-specific async data dictionary.

dialect: ClassVar[str] = 'sqlite'#: Dialect identifier. Must be defined by subclasses as a class attribute.

__init__()[source]#

get_dialect_config()[source]#

Return the dialect configuration for this data dictionary.

Return type:: DialectConfig

resolve_schema(schema)[source]#

Return a schema name using dialect defaults when missing.

Return type:: str | None

list_available_features()[source]#

List available feature flags for this dialect.

Return type:: list[str]

async get_version(driver)[source]#

Get SQLite database version information.

Parameters:: driver¶ (AiosqliteDriver) -- Async database driver instance.
Return type:: VersionInfo | None
Returns:: SQLite version information or None if detection fails.

async get_feature_flag(driver, feature)[source]#

Check if SQLite database supports a specific feature.

Parameters:

driver¶ (AiosqliteDriver) -- Async database driver instance.
feature¶ (str) -- Feature name to check.

Return type:

bool

Returns:

True if feature is supported, False otherwise.

async get_optimal_type(driver, type_category)[source]#

Get optimal SQLite type for a category.

Parameters:

driver¶ (AiosqliteDriver) -- Async database driver instance.
type_category¶ (str) -- Type category.

Return type:

str

Returns:

SQLite-specific type name.

async get_tables(driver, schema=None)[source]#

Get tables sorted by topological dependency order using SQLite catalog.

Return type:: list[TableMetadata]

async get_columns(driver, table=None, schema=None)[source]#

Get column information for a table or schema.

Return type:: list[ColumnMetadata]

async get_indexes(driver, table=None, schema=None)[source]#

Get index metadata for a table or schema.

Return type:: list[IndexMetadata]

async get_foreign_keys(driver, table=None, schema=None)[source]#

Get foreign key metadata.

Return type:: list[ForeignKeyMetadata]

Extension Settings#

Use the configuration types below in their corresponding extension_config namespace: "litestar", "events", or "adk" as supported by this adapter.

class sqlspec.adapters.aiosqlite.litestar.AiosqliteLitestarConfig[source]#

Bases: LitestarConfig

Aiosqlite-specific Litestar settings.

Use inside extension_config["litestar"] with this adapter's session store.

pragma_profile: NotRequired[bool]#

False.

Type:: Apply the extension-store PRAGMA profile. Default

pragma_overrides: NotRequired[dict[str, str | int | bool]]#: Validated PRAGMA overrides applied during schema preparation.

class sqlspec.adapters.aiosqlite.events.AiosqliteEventsConfig[source]#

Bases: EventsConfig

Aiosqlite events settings for queue storage and supported native transports.

pragma_profile: NotRequired[bool]#

False.

Type:: Apply the SQLite extension-store PRAGMA profile. Default

pragma_overrides: NotRequired[dict[str, str | int | bool]]#: Validated SQLite PRAGMA overrides applied after the optional profile.

class sqlspec.adapters.aiosqlite.adk.AiosqliteADKConfig[source]#

Bases: ADKConfig

Aiosqlite-specific ADK extension settings.

Use these keys inside extension_config["adk"] with aiosqlite ADK stores.

pragma_overrides: NotRequired[Mapping[str, str | int | bool]]#: Additional validated PRAGMA settings applied after the built-in ADK profile.

fts_tokenize: NotRequired[str]#: Optional FTS5 tokenizer spec used when memory_use_fts is enabled.

fts_detail: NotRequired[Literal['full', 'column', 'none']]#: Optional FTS5 detail mode used when memory_use_fts is enabled.