SQLite

Sync SQLite adapter using Python’s built-in sqlite3 module with thread-local connection pooling.

Configuration

class sqlspec.adapters.sqlite.SqliteConfig

Bases: SyncDatabaseConfig[Connection, SqliteConnectionPool, SqliteDriver]

SQLite configuration with thread-local connections.

driver_type

alias of SqliteDriver

connection_type

alias of Connection

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

Initialize SQLite configuration.

Parameters:

• connection_config (SqliteConnectionParams | dict[str, typing.Any] | None) – Configuration parameters including connection settings

• connection_instance (SqliteConnectionPool | None) – Pre-created pool instance

• migration_config (dict[str, typing.Any] | None) – Migration configuration

• statement_config (StatementConfig | None) – Default SQL statement configuration

• driver_features (SqliteDriverFeatures | dict[str, typing.Any] | None) – Optional driver feature configuration

• bind_key (str | None) – Optional bind key for the configuration

• extension_config (dict[str, dict[str, Any] | LitestarConfig | FastAPIConfig | StarletteConfig | FlaskConfig | ADKConfig | EventsConfig | OpenTelemetryConfig | PrometheusConfig] | None) – Extension-specific configuration (e.g., Litestar plugin settings)

• observability_config (ObservabilityConfig | None) – Adapter-level observability overrides for lifecycle hooks and observers

• **kwargs (Any) – Additional keyword arguments passed to the base configuration.

create_connection()

Get a SQLite connection from the pool.

Returns:

A connection from the pool

Return type:

SqliteConnection

provide_connection(*args, **kwargs)

Provide a SQLite connection context manager.

Parameters:

• *args (typing.Any) – Additional arguments.

• **kwargs (typing.Any) – Additional keyword arguments.

Return type:

SqliteConnectionContext

Returns:

A Sqlite connection context manager.

provide_session(*_args, statement_config=None, **_kwargs)

Provide a SQLite driver session.

Parameters:

• *_args (typing.Any) – Additional arguments.

• statement_config (StatementConfig | None) – Optional statement configuration override.

• **_kwargs (typing.Any) – Additional keyword arguments.

Return type:

SqliteSessionContext

Returns:

A Sqlite driver session context manager.

get_signature_namespace()

Get the signature namespace for SQLite types.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to types.

Driver

class sqlspec.adapters.sqlite.SqliteDriver

Bases: SyncDriverAdapterBase

SQLite driver implementation.

Provides SQL statement execution, transaction management, and result handling for SQLite databases using the standard sqlite3 module.

__init__(connection, statement_config=None, driver_features=None)

Initialize SQLite driver.

Parameters:

• connection (Connection) – SQLite database connection

• statement_config (StatementConfig | None) – Statement configuration settings

• driver_features (dict[str, typing.Any] | None) – Driver-specific feature flags

dispatch_execute(cursor, statement)

Execute single SQL statement.

Parameters:

• cursor (Cursor) – SQLite cursor object

• statement (SQL) – SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

dispatch_execute_many(cursor, statement)

Execute SQL with multiple parameter sets.

Parameters:

• cursor (Cursor) – SQLite cursor object

• statement (SQL) – SQL statement with multiple parameter sets

Return type:

ExecutionResult

Returns:

ExecutionResult with batch execution details

dispatch_execute_script(cursor, statement)

Execute SQL script with statement splitting and parameter handling.

Parameters:

• cursor (Cursor) – SQLite cursor object

• statement (SQL) – SQL statement containing multiple statements

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution details

execute_many(statement, /, parameters, *filters, statement_config=None, **kwargs)

Execute many with a SQLite thin path for simple qmark batches.

begin()

Begin a database transaction.

Raises:

SQLSpecError – If transaction cannot be started

Return type:

None

commit()

Commit the current transaction.

Raises:

SQLSpecError – If transaction cannot be committed

Return type:

None

rollback()

Rollback the current transaction.

Raises:

SQLSpecError – If transaction cannot be rolled back

Return type:

None

with_cursor(connection)

Create context manager for SQLite cursor.

Parameters:

connection (Connection) – SQLite database connection

Return type:

SqliteCursor

Returns:

Cursor context manager for safe cursor operations

handle_database_exceptions()

Handle database-specific exceptions and wrap them appropriately.

Return type:

SqliteExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility.

select_to_storage(statement, destination, /, *parameters, statement_config=None, partitioner=None, format_hint=None, telemetry=None, **kwargs)

Execute a query and write Arrow-compatible output to storage (sync).

Return type:

StorageBridgeJob

load_from_arrow(table, source, *, partitioner=None, overwrite=False, telemetry=None)

Load Arrow data into SQLite using batched inserts.

load_from_storage(table, source, *, file_format, partitioner=None, overwrite=False)

Load staged artifacts from storage into SQLite.

Return type:

StorageBridgeJob

property data_dictionary: SqliteDataDictionary

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

collect_rows(cursor, fetched)

Collect SQLite rows for the direct execution path.

Return type:

tuple[list[typing.Any], list[str], int]

resolve_rowcount(cursor)

Resolve rowcount from SQLite cursor for the direct execution path.

Return type:

int

Connection Pool

class sqlspec.adapters.sqlite.pool.SqliteConnectionPool

Bases: object

Thread-local connection manager for SQLite.

SQLite connections aren’t thread-safe, so we use thread-local storage to ensure each thread has its own connection. This is simpler and more efficient than a traditional pool for SQLite’s constraints.

__init__(connection_parameters, enable_optimizations=True, recycle_seconds=86400, health_check_interval=30.0, on_connection_create=None)

Initialize the thread-local connection manager.

Parameters:

• connection_parameters – SQLite connection parameters

• enable_optimizations – Whether to apply performance PRAGMAs

• recycle_seconds – Connection recycle time in seconds (default 24h)

• health_check_interval – Seconds of idle time before running health check

• on_connection_create – Callback executed when connection is created

get_connection()

Get a thread-local connection.

Yields:

SqliteConnection – A thread-local connection.

Return type:

Generator[Connection, None, None]

close()

Close the thread-local connection if it exists.

Return type:

None

acquire()

Acquire a thread-local connection.

Returns:

A thread-local connection

Return type:

SqliteConnection

release(connection)

Release a connection (no-op for thread-local connections).

Parameters:

connection (Connection) – The connection to release (ignored)

Return type:

None

size()

Get pool size (always 1 for thread-local).

Return type:

int

checked_out()

Get number of checked out connections (always 0).

Return type:

int