SQLite#

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

Configuration#

class sqlspec.adapters.sqlite.SqliteConfig[source]#

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)[source]#

Initialize SQLite configuration.

Parameters:
create_connection()[source]#

Open a standalone connection owned by the caller.

The connection carries the same parameters, PRAGMAs, and runtime setup the pool applies, but it is not the pool's thread-local connection, so closing it leaves the pool usable.

Returns:

A newly opened connection.

Return type:

Connection

get_signature_namespace()[source]#

Get the signature namespace for SQLite types.

Return type:

dict[str, typing.Any]

Returns:

Dictionary mapping type names to types.

Connection Parameters#

class sqlspec.adapters.sqlite.SqliteConnectionParams[source]#

Bases: TypedDict

SQLite connection parameters.

Driver Features#

class sqlspec.adapters.sqlite.SqliteDriverFeatures[source]#

Bases: TypedDict

SQLite 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 True for enhanced Python type support. Set to False only if you need pure SQLite behavior without type conversions.

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: Callback executed when a connection is created.

Receives the raw sqlite3 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 connection thread.

Each entry must include name, narg, and func.

custom_collations: Register SQL collations that compare two string values.

Each entry must include name and func.

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. trace_callback: sqlite3 trace hook run for executed statements. 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 sqlite3 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 sqlite3 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.sqlite.SqliteFunctionConfig[source]#

Bases: TypedDict

User-defined SQLite function registration.

class sqlspec.adapters.sqlite.SqliteCollationConfig[source]#

Bases: TypedDict

User-defined SQLite collation registration.

class sqlspec.adapters.sqlite.SqliteAggregateConfig[source]#

Bases: TypedDict

User-defined SQLite aggregate registration.

Driver#

class sqlspec.adapters.sqlite.SqliteDriver[source]#

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)[source]#

Initialize SQLite driver.

Parameters:
dispatch_execute(cursor, statement)[source]#

Execute single SQL statement.

Parameters:

  • cursor (Any) -- SQLite cursor object

  • statement (SQL) -- SQL statement to execute

Return type:

ExecutionResult

Returns:

ExecutionResult with statement execution details

dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets.

Parameters:

  • cursor (Any) -- 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)[source]#

Execute SQL script with statement splitting and parameter handling.

Parameters:

  • cursor (Any) -- 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)[source]#

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

begin(mode=None)[source]#

Begin a database transaction.

Parameters:

mode (Optional[Literal['DEFERRED', 'IMMEDIATE', 'EXCLUSIVE']]) -- Transaction lock mode (DEFERRED, IMMEDIATE, or EXCLUSIVE). Defaults to configured driver feature or SQLite default (DEFERRED).

Raises:

SQLSpecError -- If transaction cannot be started

Return type:

None

commit()[source]#

Commit the current transaction.

Raises:

SQLSpecError -- If transaction cannot be committed

Return type:

None

rollback()[source]#

Rollback the current transaction.

Raises:

SQLSpecError -- If transaction cannot be rolled back

Return type:

None

with_cursor(connection)[source]#

Create context manager for SQLite cursor.

Parameters:

connection (Connection) -- SQLite database connection

Return type:

SqliteCursor

Returns:

Cursor context manager for safe cursor operations

dispatch_select_stream(statement, chunk_size)[source]#

Return a native SQLite row stream backed by chunked fetchmany.

Return type:

Optional[SyncRowStream[dict[str, typing.Any]]]

handle_database_exceptions()[source]#

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)[source]#

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

Return type:

StorageBridgeJob

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

Load Arrow data into SQLite using chunked batched inserts.

Return type:

StorageBridgeJob

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

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)[source]#

Collect SQLite rows for the direct execution path.

Return type:

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

resolve_rowcount(cursor)[source]#

Resolve rowcount from SQLite cursor for the direct execution path.

Return type:

int

Connection Pool#

class sqlspec.adapters.sqlite.pool.SqliteConnectionPool[source]#

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, enable_foreign_keys=False, recycle_seconds=86400, health_check_interval=30.0, on_connection_create=None, runtime_setup=None)[source]#

Initialize the thread-local connection manager.

Parameters:

  • connection_parameters (dict[str, typing.Any]) -- SQLite connection parameters

  • enable_optimizations (bool) -- Whether to apply performance PRAGMAs

  • enable_foreign_keys (bool) -- Whether to enable foreign-key enforcement

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

  • health_check_interval (float) -- Seconds of idle time before running health check

  • on_connection_create (typing.Callable[[<class 'sqlite3.Connection'>], None] | None) -- Callback executed when connection is created

  • runtime_setup (dict[str, typing.Any] | None) -- Runtime feature configuration applied after internal PRAGMAs

new_connection()[source]#

Create a standalone connection configured like a pooled one.

The result is owned by the caller: it is not thread-local and is not tracked for pool shutdown.

Returns:

A newly opened, fully configured connection.

Return type:

Connection

get_connection()[source]#

Get a thread-local connection.

Yields:

SqliteConnection -- A thread-local connection.

close()[source]#

Close every connection this pool opened, on any thread.

Return type:

None

acquire()[source]#

Acquire a thread-local connection.

Returns:

A thread-local connection

Return type:

Connection

release(connection)[source]#

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

Parameters:

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

Return type:

None

size()[source]#

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

Return type:

int

checked_out()[source]#

Get number of checked out connections (always 0).

Return type:

int

Data Dictionary#

class sqlspec.adapters.sqlite.data_dictionary.SqliteDataDictionary[source]#

Bases: SyncDataDictionaryBase

SQLite-specific sync data dictionary.

dialect: ClassVar[str] = 'sqlite'#

Dialect identifier. Must be defined by subclasses as a class attribute.

__init__()[source]#
resolve_schema(schema)[source]#

Return a schema name using dialect defaults when missing.

Return type:

str | None

get_version(driver)[source]#

Get SQLite database version information.

Parameters:

driver (SqliteDriver) -- Sync database driver instance.

Return type:

VersionInfo | None

Returns:

SQLite version information or None if detection fails.

get_feature_flag(driver, feature)[source]#

Check if SQLite database supports a specific feature.

Parameters:

  • driver (SqliteDriver) -- Sync database driver instance.

  • feature (str) -- Feature name to check.

Return type:

bool

Returns:

True if feature is supported, False otherwise.

get_optimal_type(driver, type_category)[source]#

Get optimal SQLite type for a category.

Parameters:

  • driver (SqliteDriver) -- Sync database driver instance.

  • type_category (str) -- Type category.

Return type:

str

Returns:

SQLite-specific type name.

list_available_features()[source]#

List available feature flags for this dialect.

Return type:

list[str]

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

Get tables sorted by topological dependency order using SQLite catalog.

Return type:

list[TableMetadata]

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

Get column information for a table or schema.

Return type:

list[ColumnMetadata]

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

Get index metadata for a table or schema.

Return type:

list[IndexMetadata]

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.sqlite.litestar.SqliteLitestarConfig[source]#

Bases: LitestarConfig

Sqlite-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.sqlite.events.SqliteEventsConfig[source]#

Bases: EventsConfig

Sqlite 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.sqlite.adk.SqliteADKConfig[source]#

Bases: ADKConfig

SQLite-specific ADK extension settings.

Use these keys inside extension_config["adk"] with SQLite 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.