Driver#

The driver module defines sync and async driver adapters, transaction helpers, and the shared data dictionary mixins.

Example#

driver usage#
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

db_path = tmp_path / "driver_api.db"
spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": str(db_path)}))

with spec.provide_session(config) as session:
    session.execute("create table if not exists users (id integer primary key, name text)")
    session.execute("insert into users (name) values ('Ada')")
    row = session.select_one_or_none("select name from users where id = ?", 1)
    print(row)

Transaction Blocks#

transaction() wraps a block in a transaction on the driver's connection. Entering the block calls begin() and yields the same driver. A normal exit commits; an exception rolls back and propagates to the caller. If the commit itself fails, the block attempts a rollback and then raises the commit error.

The block calls the adapter's own begin(), commit(), and rollback(), so it follows each database's transaction model. When the connection already has an open transaction, whether started by begin() or implicitly by an earlier statement as SQLite does in its default mode or a connection with autocommit disabled does, the block joins it instead of calling begin(). Exiting the block commits or rolls back that whole transaction, including work done before the block.

async with config.provide_session() as session:
    async with session.transaction():
        await session.execute("INSERT INTO users (name) VALUES (:name)", name="Ada")
        await session.execute("INSERT INTO audit (action) VALUES (:action)", action="user-created")

with config.provide_session() as session:
    with session.transaction():
        session.execute("INSERT INTO users (name) VALUES (:name)", name="Ada")
        session.execute("INSERT INTO audit (action) VALUES (:action)", action="user-created")

Nested blocks#

A transaction() block entered inside another transaction() block on the same driver, or inside a service's begin_transaction() block that uses the driver, does not begin or commit. It runs in a savepoint instead: a normal exit releases the savepoint, and an exception rolls back to it and propagates. The enclosing block stays open and decides whether the work is committed. A service begin_transaction() block inside a transaction() block nests the same way.

from sqlspec.exceptions import UniqueViolationError

with session.transaction():
    session.execute("INSERT INTO users (name) VALUES (:name)", name="Ada")
    try:
        with session.transaction():
            session.execute("INSERT INTO users (name) VALUES (:name)", name="Ada")
    except UniqueViolationError:
        pass

Nesting needs savepoint support. When the adapter cannot create a savepoint, entering a nested block raises ImproperConfigurationError and the enclosing block stays usable. DuckDB and ADBC connections to DuckDB, BigQuery, or Snowflake report missing savepoint support. BigQuery has no transactions, so its begin(), commit(), and rollback() do nothing, and Spanner commits or rolls back only sessions opened for writes; neither supports savepoints, so nested blocks are not supported on either.

Isolation settings#

transaction() takes no isolation-level argument. Apply isolation or other transaction settings with execute_script as the first statement inside the block, using the syntax your database supports:

async with session.transaction():
    await session.execute_script("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
    await session.execute(
        "UPDATE accounts SET balance = balance - :amount WHERE id = :id",
        amount=10,
        id=1,
    )

Driver Adapter Protocol and Base Classes#

SQLSpec does not define standalone DriverProtocol, AsyncDriverProtocol, or SessionProtocol classes. Instead, database drivers and sessions are instances of SyncDriverAdapterBase or AsyncDriverAdapterBase. The type alias DriverAdapterProtocol unifies synchronous and asynchronous driver adapters for generic annotations.

sqlspec.driver.DriverAdapterProtocol = sqlspec.driver._sync.SyncDriverAdapterBase | sqlspec.driver._async.AsyncDriverAdapterBase#

Represent a PEP 604 union type

E.g. for int | str

Synchronous Driver Adapter#

class sqlspec.driver.SyncDriverAdapterBase[source]#

Bases: CommonDriverAttributesMixin

Base class for synchronous database drivers.

This class includes flattened storage and SQL translation methods that were previously in StorageDriverMixin and SQLTranslatorMixin. The flattening eliminates cross-trait attribute access that caused mypyc segmentation faults.

Method Organization:

  1. Core dispatch methods (the execution engine)

  2. Transaction management (abstract methods)

  3. Public API - execution methods

  4. Public API - query methods (select/fetch variants)

  5. Arrow API methods

  6. Stack execution

  7. Storage API methods

  8. Utility methods

  9. Private/internal methods

dialect: DialectType | None = None#
abstract property data_dictionary: SyncDataDictionaryBase#

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

set_migration_session_schema(schema)[source]#

Set the default schema for migration SQL when supported.

Parameters:

schema (str) -- Schema requested for the current migration session.

Return type:

None

set_migration_non_transactional_schema(schema)[source]#

Set the default schema for non-transactional migration SQL when supported.

Parameters:

schema (str) -- Schema requested for the current migration session.

Return type:

None

reset_migration_session_schema()[source]#

Reset migration schema state after a non-transactional migration.

Return type:

None

has_schema(schema)[source]#

Return whether the schema exists for migration validation.

Parameters:

schema (str) -- Schema name to validate.

Return type:

bool

Returns:

True when the adapter does not provide schema validation.

final dispatch_statement_execution(statement, connection)[source]#

Central execution dispatcher using the Template Method Pattern.

Parameters:

  • statement (SQL) -- The SQL statement to execute

  • connection (typing.Any) -- The database connection to use

Return type:

SQLResult

Returns:

The result of the SQL execution

abstractmethod dispatch_execute(cursor, statement)[source]#

Execute a single SQL statement.

Must be implemented by each driver for database-specific execution logic.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data

abstractmethod dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets (executemany).

Must be implemented by each driver for database-specific executemany logic.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data for the many operation

dispatch_execute_script(cursor, statement)[source]#

Execute a SQL script containing multiple statements.

Default implementation splits the script and executes statements individually. Drivers can override for database-specific script execution methods.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution data including statement counts

dispatch_special_handling(cursor, statement)[source]#

Hook for database-specific special operations.

This method is called first in dispatch_statement_execution() to allow drivers to handle special operations that don't follow the standard SQL execution pattern.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement to analyze

Return type:

SQLResult | None

Returns:

SQLResult if the special operation was handled and completed, None if standard execution should proceed

collect_rows(cursor, fetched)[source]#

Collect rows from cursor after fetchall for the direct execution path.

Adapters should override this method to provide optimized row collection that bypasses full dispatch_execute overhead.

Parameters:

  • cursor (Any) -- Database cursor with description metadata.

  • fetched (list[typing.Any]) -- Rows returned from cursor.fetchall().

Return type:

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

Returns:

Tuple of (data, column_names, row_count).

Raises:

NotImplementedError -- If the adapter does not implement this method.

resolve_rowcount(cursor)[source]#

Resolve the number of affected rows from cursor for the direct execution path.

Adapters should override this method to provide optimized rowcount resolution that bypasses full dispatch_execute overhead.

Parameters:

cursor (Any) -- Database cursor with rowcount metadata.

Return type:

int

Returns:

Number of affected rows, or 0 when unknown.

Raises:

NotImplementedError -- If the adapter does not implement this method.

abstractmethod begin()[source]#

Begin a database transaction on the current connection.

Return type:

None

abstractmethod commit()[source]#

Commit the current transaction on the current connection.

Return type:

None

abstractmethod rollback()[source]#

Rollback the current transaction on the current connection.

Return type:

None

create_savepoint(name)[source]#

Create a savepoint within the current transaction.

Return type:

None

release_savepoint(name)[source]#

Release a previously created savepoint.

Return type:

None

rollback_to_savepoint(name)[source]#

Roll back the current transaction to a previously created savepoint.

Return type:

None

transaction()[source]#

Return a context manager that wraps a block in a transaction.

Entering the block calls begin() and yields this driver. A normal exit calls commit(); an exception calls rollback() and propagates. A failed commit is followed by a rollback attempt before the commit error propagates. When the connection already has an open transaction, the block joins it instead of calling begin() and commits it on exit. Inside another transaction() or service begin_transaction() block on this driver, the block runs in a savepoint instead and leaves the outer transaction open. Isolation settings are applied with execute_script inside the block.

Example

with session.transaction():
    session.execute(
        "INSERT INTO items (id) VALUES (:id)", id=1
    )
Return type:

_SyncDriverTransaction[Self]

Returns:

A context manager yielding this driver.

abstractmethod with_cursor(connection)[source]#

Create and return a context manager for cursor acquisition and cleanup.

Returns a context manager that yields a cursor for database operations. Concrete implementations handle database-specific cursor creation and cleanup.

Return type:

Any

abstractmethod handle_database_exceptions()[source]#

Handle database-specific exceptions and wrap them appropriately.

Return type:

SyncExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility. The handler stores mapped exceptions in pending_exception rather than raising from __exit__ to avoid ABI boundary violations.

execute(statement, /, *parameters, statement_config=None, **kwargs)[source]#

Execute a statement with parameter handling.

execute_many(statement, /, parameters, *filters, statement_config=None, **kwargs)[source]#

Execute statement multiple times with different parameters.

Parameters passed will be used as the batch execution sequence.

execute_script(statement, /, *parameters, statement_config=None, **kwargs)[source]#

Execute a multi-statement script.

By default, validates each statement and logs warnings for dangerous operations. Use suppress_warnings=True for migrations and admin scripts.

select(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return all rows.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → list[SchemaT]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → list[dict[str, Any]]

fetch(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return all rows.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → list[SchemaT]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → list[dict[str, Any]]

This is an alias for select() provided for users familiar with asyncpg's fetch() naming convention.

See also

select(): Primary method with identical behavior

select_one(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return exactly one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any]

Raises an exception if no rows or more than one row is returned.

fetch_one(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return exactly one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any]

This is an alias for select_one() provided for users familiar with asyncpg's fetch_one() naming convention.

Raises an exception if no rows or more than one row is returned.

See also

select_one(): Primary method with identical behavior

select_one_or_none(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return at most one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any] | None

Returns None if no rows are found. Raises MultipleResultsFoundError if more than one row is returned. Any database or SQL execution errors raised by the driver are propagated unchanged.

fetch_one_or_none(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return at most one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any] | None

This is an alias for select_one_or_none() provided for users familiar with asyncpg's fetch_one_or_none() naming convention.

Returns None if no rows are found. Raises an exception if more than one row is returned.

See also

select_one_or_none(): Primary method with identical behavior

select_value(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

Expects exactly one row with one column. Raises an exception if no rows or more than one row/column is returned.

Parameters:

  • statement -- SQL statement or query builder to execute.

  • *parameters -- Positional parameters for the statement.

  • value_type -- Optional type to convert the result to. When provided, the return value is converted to this type and the return type is narrowed for type checkers. Supports int, float, str, bool, datetime, date, time, Decimal, UUID, Path, dict, and list.

  • statement_config -- Optional statement configuration.

  • **kwargs -- Additional keyword arguments.

Returns:

The scalar value, optionally converted to the specified type.

fetch_value(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

This is an alias for select_value() provided for users familiar with asyncpg's fetch_value() naming convention.

Expects exactly one row with one column. Raises an exception if no rows or more than one row/column is returned.

See also

select_value(): Primary method with identical behavior

select_value_or_none(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value or None.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

Returns None if no rows are found. Expects at most one row with one column. Raises an exception if more than one row is returned.

Parameters:

  • statement -- SQL statement or query builder to execute.

  • *parameters -- Positional parameters for the statement.

  • value_type -- Optional type to convert the result to. When provided, the return value is converted to this type and the return type is narrowed to T | None for type checkers. Supports int, float, str, bool, datetime, date, time, Decimal, UUID, Path, dict, and list.

  • statement_config -- Optional statement configuration.

  • **kwargs -- Additional keyword arguments.

Returns:

The scalar value (optionally converted), or None if no rows found.

fetch_value_or_none(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value or None.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

This is an alias for select_value_or_none() provided for users familiar with asyncpg's fetch_value_or_none() naming convention.

Returns None if no rows are found. Expects at most one row with one column. Raises an exception if more than one row is returned.

See also

select_value_or_none(): Primary method with identical behavior

select_with_total(statement, /, *parameters, schema_type=None, statement_config=None, count_with_window=False, **kwargs)[source]#

Execute a select statement and return both the data and total count.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[SchemaT], int]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[dict[str, Any]], int]

This method is designed for pagination scenarios where you need both the current page of data and the total number of rows that match the query.

Parameters:

  • statement -- The SQL statement, QueryBuilder, or raw SQL string

  • *parameters -- Parameters for the SQL statement

  • schema_type -- Optional schema type for data transformation

  • statement_config -- Optional SQL configuration

  • count_with_window -- If True, use a single query with COUNT(*) OVER() window function instead of two separate queries. This can be more efficient for some databases but adds a column to each row. Default False.

  • **kwargs -- Additional keyword arguments

Returns: A tuple containing:

  • List of data rows (transformed by schema_type if provided)

  • Total count of rows matching the query (ignoring LIMIT/OFFSET)

fetch_with_total(statement, /, *parameters, schema_type=None, statement_config=None, count_with_window=False, **kwargs)[source]#

Execute a select statement and return both the data and total count.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[SchemaT], int]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[dict[str, Any]], int]

This is an alias for select_with_total() provided for users familiar with asyncpg's fetch() naming convention.

This method is designed for pagination scenarios where you need both the current page of data and the total number of rows that match the query.

See also

select_with_total(): Primary method with identical behavior and full documentation

select_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)[source]#

Execute query and return results as Apache Arrow format.

This base implementation uses the conversion path: execute() → dict → Arrow. Adapters with native Arrow support (ADBC, DuckDB, BigQuery) override this method to use zero-copy native paths for 5-10x performance improvement.

Parameters:

  • statement -- SQL query string, Statement, or QueryBuilder

  • *parameters -- Query parameters (same format as execute()/select())

  • statement_config -- Optional statement configuration override

  • return_format -- "table" for pyarrow.Table (default), "batch" for single RecordBatch, "batches" for iterator of RecordBatches, "reader" for RecordBatchReader

  • native_only -- If True, raise error if native Arrow unavailable (default: False)

  • batch_size -- Rows per batch for "batch"/"batches" format (default: None = all rows)

  • arrow_schema -- Optional pyarrow.Schema for type casting

  • **kwargs -- Additional keyword arguments

Returns:

ArrowResult containing pyarrow.Table, RecordBatchReader, or RecordBatches

Raises:

ImproperConfigurationError -- If native_only=True and adapter doesn't support native Arrow

fetch_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)[source]#

Execute query and return results as Apache Arrow format.

This is an alias for select_to_arrow() provided for users familiar with asyncpg's fetch() naming convention.

See also

select_to_arrow(): Primary method with identical behavior and full documentation

select_stream(statement, /, *parameters, schema_type=None, statement_config=None, chunk_size=1000, native_only=False, **kwargs)[source]#

Execute a query and stream rows in chunks.

Overloads:

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → SyncRowStream[SchemaT]

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → SyncRowStream[dict[str, Any]]

fetch_stream(statement, /, *parameters, schema_type=None, statement_config=None, chunk_size=1000, native_only=False, **kwargs)[source]#

Execute a query and stream rows in chunks.

Overloads:

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → SyncRowStream[SchemaT]

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → SyncRowStream[dict[str, Any]]

This is an alias for select_stream() provided for users familiar with asyncpg's fetch() naming convention.

See also

select_stream(): Primary method with identical behavior and full documentation

dispatch_select_stream(statement, chunk_size)[source]#

Adapter hook returning a native row stream, or None when unsupported.

Return type:

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

execute_stack(stack, *, continue_on_error=False)[source]#

Execute a StatementStack sequentially using the adapter's primitives.

Return type:

tuple[StackResult, ...]

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

Stream a SELECT statement directly into storage.

Parameters:

  • statement -- SQL statement to execute.

  • destination -- Storage destination path.

  • parameters -- Query parameters.

  • statement_config -- Optional statement configuration.

  • partitioner -- Optional partitioner configuration.

  • format_hint -- Optional format hint for storage.

  • telemetry -- Optional telemetry dict to merge.

Returns:

StorageBridgeJob with execution telemetry.

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

Load Arrow data into the target table.

Parameters:

  • table (str) -- Target table name.

  • source (Union[ArrowResult, typing.Any]) -- Arrow data source.

  • partitioner (dict[str, object] | None) -- Optional partitioner configuration.

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

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

Load artifacts from storage into the target table.

Parameters:

  • table (str) -- Target table name.

  • source (str | Path) -- Storage source path.

  • file_format (Literal['jsonl', 'json', 'parquet', 'arrow-ipc', 'csv']) -- File format of source.

  • partitioner (dict[str, object] | None) -- Optional partitioner configuration.

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

load_from_records(table, records, *, columns=None, overwrite=False)[source]#

Load in-memory dict or positional records into the target table.

Records are normalized into an Arrow table and routed through the adapter's native load_from_arrow path (COPY, executemany, mutations, etc.). Dict records derive their columns from the keys; positional records require columns.

Parameters:

  • table (str) -- Target table name.

  • records (Sequence[Mapping[str, typing.Any]] | Sequence[Sequence[typing.Any]]) -- Mapping records, or positional sequences with columns.

  • columns (list[str] | None) -- Column names (required for positional records).

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

convert_to_dialect(statement, to_dialect=None, pretty=True)[source]#

Convert a statement to a target SQL dialect.

Parameters:

  • statement (str | Expr | SQL) -- SQL statement to convert.

  • to_dialect (Optional[sqlglot.dialects.dialect.DialectType]) -- Target dialect (defaults to current dialect).

  • pretty (bool) -- Whether to format the output SQL.

Return type:

str

Returns:

SQL string in target dialect.

Asynchronous Driver Adapter#

class sqlspec.driver.AsyncDriverAdapterBase[source]#

Bases: CommonDriverAttributesMixin

Base class for asynchronous database drivers.

This class includes flattened storage and SQL translation methods that were previously in StorageDriverMixin and SQLTranslatorMixin. The flattening eliminates cross-trait attribute access that caused mypyc segmentation faults.

Method Organization:

  1. Core dispatch methods (the execution engine)

  2. Transaction management (abstract methods)

  3. Public API - execution methods

  4. Public API - query methods (select/fetch variants)

  5. Arrow API methods

  6. Stack execution

  7. Storage API methods

  8. Utility methods

  9. Private/internal methods

dialect: DialectType | None = None#
property is_async: bool#

Return whether the driver executes asynchronously.

Returns:

True for async drivers.

abstract property data_dictionary: AsyncDataDictionaryBase#

Get the data dictionary for this driver.

Returns:

Data dictionary instance for metadata queries

async set_migration_session_schema(schema)[source]#

Set the default schema for migration SQL when supported.

Parameters:

schema (str) -- Schema requested for the current migration session.

Return type:

None

async set_migration_non_transactional_schema(schema)[source]#

Set the default schema for non-transactional migration SQL when supported.

Parameters:

schema (str) -- Schema requested for the current migration session.

Return type:

None

async reset_migration_session_schema()[source]#

Reset migration schema state after a non-transactional migration.

Return type:

None

async has_schema(schema)[source]#

Return whether the schema exists for migration validation.

Parameters:

schema (str) -- Schema name to validate.

Return type:

bool

Returns:

True when the adapter does not provide schema validation.

final async dispatch_statement_execution(statement, connection)[source]#

Central execution dispatcher using the Template Method Pattern.

Parameters:

  • statement (SQL) -- The SQL statement to execute

  • connection (typing.Any) -- The database connection to use

Return type:

SQLResult

Returns:

The result of the SQL execution

abstractmethod async dispatch_execute(cursor, statement)[source]#

Execute a single SQL statement.

Must be implemented by each driver for database-specific execution logic.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data

abstractmethod async dispatch_execute_many(cursor, statement)[source]#

Execute SQL with multiple parameter sets (executemany).

Must be implemented by each driver for database-specific executemany logic.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with execution data for the many operation

async dispatch_execute_script(cursor, statement)[source]#

Execute a SQL script containing multiple statements.

Default implementation splits the script and executes statements individually. Drivers can override for database-specific script execution methods.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement object with all necessary data and configuration

Return type:

ExecutionResult

Returns:

ExecutionResult with script execution data including statement counts

async dispatch_special_handling(cursor, statement)[source]#

Hook for database-specific special operations.

This method is called first in dispatch_statement_execution() to allow drivers to handle special operations that don't follow the standard SQL execution pattern.

Parameters:

  • cursor (Any) -- Database cursor/connection object

  • statement (SQL) -- SQL statement to analyze

Return type:

SQLResult | None

Returns:

SQLResult if the special operation was handled and completed, None if standard execution should proceed

collect_rows(cursor, fetched)[source]#

Collect rows from cursor after fetchall for the direct execution path.

Adapters should override this method to provide optimized row collection that bypasses full dispatch_execute overhead.

Parameters:

  • cursor (Any) -- Database cursor with description metadata.

  • fetched (list[typing.Any]) -- Rows returned from cursor.fetchall().

Return type:

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

Returns:

Tuple of (data, column_names, row_count).

Raises:

NotImplementedError -- If the adapter does not implement this method.

resolve_rowcount(cursor)[source]#

Resolve the number of affected rows from cursor for the direct execution path.

Adapters should override this method to provide optimized rowcount resolution that bypasses full dispatch_execute overhead.

Parameters:

cursor (Any) -- Database cursor with rowcount metadata.

Return type:

int

Returns:

Number of affected rows, or 0 when unknown.

Raises:

NotImplementedError -- If the adapter does not implement this method.

abstractmethod async begin()[source]#

Begin a database transaction on the current connection.

Return type:

None

abstractmethod async commit()[source]#

Commit the current transaction on the current connection.

Return type:

None

abstractmethod async rollback()[source]#

Rollback the current transaction on the current connection.

Return type:

None

async create_savepoint(name)[source]#

Create a savepoint within the current transaction.

Return type:

None

async release_savepoint(name)[source]#

Release a previously created savepoint.

Return type:

None

async rollback_to_savepoint(name)[source]#

Roll back the current transaction to a previously created savepoint.

Return type:

None

transaction()[source]#

Return a context manager that wraps a block in a transaction.

Entering the block calls begin() and yields this driver. A normal exit calls commit(); an exception calls rollback() and propagates. A failed commit is followed by a rollback attempt before the commit error propagates. When the connection already has an open transaction, the block joins it instead of calling begin() and commits it on exit. Inside another transaction() or service begin_transaction() block on this driver, the block runs in a savepoint instead and leaves the outer transaction open. Isolation settings are applied with execute_script inside the block.

Example

async with session.transaction():
    await session.execute(
        "INSERT INTO items (id) VALUES (:id)", id=1
    )
Return type:

_AsyncDriverTransaction[Self]

Returns:

A context manager yielding this driver.

abstractmethod with_cursor(connection)[source]#

Create and return an async context manager for cursor acquisition and cleanup.

Returns an async context manager that yields a cursor for database operations. Concrete implementations handle database-specific cursor creation and cleanup.

Return type:

Any

abstractmethod handle_database_exceptions()[source]#

Handle database-specific exceptions and wrap them appropriately.

Return type:

AsyncExceptionHandler

Returns:

Exception handler with deferred exception pattern for mypyc compatibility. The handler stores mapped exceptions in pending_exception rather than raising from __aexit__ to avoid ABI boundary violations.

async execute(statement, /, *parameters, statement_config=None, **kwargs)[source]#

Execute a statement with parameter handling.

async execute_many(statement, /, parameters, *filters, statement_config=None, **kwargs)[source]#

Execute statement multiple times with different parameters.

Parameters passed will be used as the batch execution sequence.

async execute_script(statement, /, *parameters, statement_config=None, **kwargs)[source]#

Execute a multi-statement script.

By default, validates each statement and logs warnings for dangerous operations. Use suppress_warnings=True for migrations and admin scripts.

async select(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return all rows.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → list[SchemaT]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → list[dict[str, Any]]

async fetch(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return all rows.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → list[SchemaT]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → list[dict[str, Any]]

This is an alias for select() provided for users familiar with asyncpg's fetch() naming convention.

See also

select(): Primary method with identical behavior

async select_one(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return exactly one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any]

Raises an exception if no rows or more than one row is returned.

async fetch_one(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return exactly one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any]

This is an alias for select_one() provided for users familiar with asyncpg's fetch_one() naming convention.

Raises an exception if no rows or more than one row is returned.

See also

select_one(): Primary method with identical behavior

async select_one_or_none(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return at most one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any] | None

Returns None if no rows are found. Raises MultipleResultsFoundError if more than one row is returned. Any database or SQL execution errors raised by the driver are propagated unchanged.

async fetch_one_or_none(statement, /, *parameters, schema_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return at most one row.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), kwargs (Any) → SchemaT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), kwargs (Any) → dict[str, Any] | None

This is an alias for select_one_or_none() provided for users familiar with asyncpg's fetch_one_or_none() naming convention.

Returns None if no rows are found. Raises an exception if more than one row is returned.

See also

select_one_or_none(): Primary method with identical behavior

async select_value(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

Expects exactly one row with one column. Raises an exception if no rows or more than one row/column is returned.

Parameters:

  • statement -- SQL statement or query builder to execute.

  • *parameters -- Positional parameters for the statement.

  • value_type -- Optional type to convert the result to. When provided, the return value is converted to this type and the return type is narrowed for type checkers. Supports int, float, str, bool, datetime, date, time, Decimal, UUID, Path, dict, and list.

  • statement_config -- Optional statement configuration.

  • **kwargs -- Additional keyword arguments.

Returns:

The scalar value, optionally converted to the specified type

async fetch_value(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

This is an alias for select_value() provided for users familiar with asyncpg's fetch_value() naming convention.

Expects exactly one row with one column. Raises an exception if no rows or more than one row/column is returned.

See also

select_value(): Primary method with identical behavior

async select_value_or_none(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value or None.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

Returns None if no rows are found. Expects at most one row with one column. Raises an exception if more than one row is returned.

Parameters:

  • statement -- SQL statement or query builder to execute.

  • *parameters -- Positional parameters for the statement.

  • value_type -- Optional type to convert the result to. When provided, the return value is converted to this type and the return type is narrowed to T | None for type checkers. Supports int, float, str, bool, datetime, date, time, Decimal, UUID, Path, dict, and list.

  • statement_config -- Optional statement configuration.

  • **kwargs -- Additional keyword arguments.

Returns:

The scalar value (optionally converted), or None if no rows found.

async fetch_value_or_none(statement, /, *parameters, value_type=None, statement_config=None, **kwargs)[source]#

Execute a select statement and return a single scalar value or None.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (type[ValueT]), statement_config (StatementConfig | None), kwargs (Any) → ValueT | None

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), value_type (None), statement_config (StatementConfig | None), kwargs (Any) → Any

This is an alias for select_value_or_none() provided for users familiar with asyncpg's fetch_value_or_none() naming convention.

Returns None if no rows are found. Expects at most one row with one column. Raises an exception if more than one row is returned.

See also

select_value_or_none(): Primary method with identical behavior

async select_with_total(statement, /, *parameters, schema_type=None, statement_config=None, count_with_window=False, **kwargs)[source]#

Execute a select statement and return both the data and total count.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[SchemaT], int]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[dict[str, Any]], int]

This method is designed for pagination scenarios where you need both the current page of data and the total number of rows that match the query.

Parameters:

  • statement -- The SQL statement, QueryBuilder, or raw SQL string

  • *parameters -- Parameters for the SQL statement

  • schema_type -- Optional schema type for data transformation

  • statement_config -- Optional SQL configuration

  • count_with_window -- If True, use a single query with COUNT(*) OVER() window function instead of two separate queries. This can be more efficient for some databases but adds a column to each row. Default False.

  • **kwargs -- Additional keyword arguments

Returns: A tuple containing:

  • List of data rows (transformed by schema_type if provided)

  • Total count of rows matching the query (ignoring LIMIT/OFFSET)

async fetch_with_total(statement, /, *parameters, schema_type=None, statement_config=None, count_with_window=False, **kwargs)[source]#

Execute a select statement and return both the data and total count.

Overloads:

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[SchemaT], int]

  • self, statement (Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), count_with_window (bool), kwargs (Any) → tuple[list[dict[str, Any]], int]

This is an alias for select_with_total() provided for users familiar with asyncpg's fetch() naming convention.

This method is designed for pagination scenarios where you need both the current page of data and the total number of rows that match the query.

See also

select_with_total(): Primary method with identical behavior and full documentation

async select_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)[source]#

Execute query and return results as Apache Arrow format (async).

This base implementation uses the conversion path: execute() → dict → Arrow. Adapters with native Arrow support (ADBC, DuckDB, BigQuery) override this method to use zero-copy native paths for 5-10x performance improvement.

Parameters:

  • statement -- SQL query string, Statement, or QueryBuilder

  • *parameters -- Query parameters (same format as execute()/select())

  • statement_config -- Optional statement configuration override

  • return_format -- "table" for pyarrow.Table (default), "batch" for single RecordBatch, "batches" for iterator of RecordBatches, "reader" for RecordBatchReader

  • native_only -- If True, raise error if native Arrow unavailable (default: False)

  • batch_size -- Rows per batch for "batch"/"batches" format (default: None = all rows)

  • arrow_schema -- Optional pyarrow.Schema for type casting

  • **kwargs -- Additional keyword arguments

Returns:

ArrowResult containing pyarrow.Table, RecordBatchReader, or RecordBatches

Raises:

ImproperConfigurationError -- If native_only=True and adapter doesn't support native Arrow

async fetch_to_arrow(statement, /, *parameters, statement_config=None, return_format='table', native_only=False, batch_size=None, arrow_schema=None, **kwargs)[source]#

Execute query and return results as Apache Arrow format (async).

This is an alias for select_to_arrow() provided for users familiar with asyncpg's fetch() naming convention.

See also

select_to_arrow(): Primary method with identical behavior and full documentation

select_stream(statement, /, *parameters, schema_type=None, statement_config=None, chunk_size=1000, native_only=False, **kwargs)[source]#

Execute a query and stream rows in chunks.

Overloads:

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → AsyncRowStream[SchemaT]

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → AsyncRowStream[dict[str, Any]]

fetch_stream(statement, /, *parameters, schema_type=None, statement_config=None, chunk_size=1000, native_only=False, **kwargs)[source]#

Execute a query and stream rows in chunks.

Overloads:

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (type[SchemaT]), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → AsyncRowStream[SchemaT]

  • self, statement (SQL | Statement | QueryBuilder), parameters (StatementParameters | StatementFilter), schema_type (None), statement_config (StatementConfig | None), chunk_size (int), native_only (bool), kwargs (Any) → AsyncRowStream[dict[str, Any]]

This is an alias for select_stream() provided for users familiar with asyncpg's fetch() naming convention.

See also

select_stream(): Primary method with identical behavior and full documentation

dispatch_select_stream(statement, chunk_size)[source]#

Adapter hook returning a native row stream, or None when unsupported.

Return type:

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

async execute_stack(stack, *, continue_on_error=False)[source]#

Execute a StatementStack sequentially using the adapter's primitives.

Return type:

tuple[StackResult, ...]

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

Stream a SELECT statement directly into storage.

Parameters:

  • statement -- SQL statement to execute.

  • destination -- Storage destination path.

  • parameters -- Query parameters.

  • statement_config -- Optional statement configuration.

  • partitioner -- Optional partitioner configuration.

  • format_hint -- Optional format hint for storage.

  • telemetry -- Optional telemetry dict to merge.

Returns:

StorageBridgeJob with execution telemetry.

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

Load Arrow data into the target table.

Parameters:

  • table (str) -- Target table name.

  • source (Union[ArrowResult, typing.Any]) -- Arrow data source.

  • partitioner (dict[str, object] | None) -- Optional partitioner configuration.

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

Raises:

NotImplementedError -- If not implemented.

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

Load artifacts from storage into the target table.

Parameters:

  • table (str) -- Target table name.

  • source (str | Path) -- Storage source path.

  • file_format (Literal['jsonl', 'json', 'parquet', 'arrow-ipc', 'csv']) -- File format of source.

  • partitioner (dict[str, object] | None) -- Optional partitioner configuration.

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

async load_from_records(table, records, *, columns=None, overwrite=False)[source]#

Load in-memory dict or positional records into the target table.

Records are normalized into an Arrow table and routed through the adapter's native load_from_arrow path (COPY, executemany, mutations, etc.). Dict records derive their columns from the keys; positional records require columns.

Parameters:

  • table (str) -- Target table name.

  • records (Sequence[Mapping[str, typing.Any]] | Sequence[Sequence[typing.Any]]) -- Mapping records, or positional sequences with columns.

  • columns (list[str] | None) -- Column names (required for positional records).

  • overwrite (bool) -- Whether to overwrite existing data.

Return type:

StorageBridgeJob

Returns:

StorageBridgeJob with execution telemetry.

convert_to_dialect(statement, to_dialect=None, pretty=True)[source]#

Convert a statement to a target SQL dialect.

Parameters:

  • statement (str | Expr | SQL) -- SQL statement to convert.

  • to_dialect (Optional[sqlglot.dialects.dialect.DialectType]) -- Target dialect (defaults to current dialect).

  • pretty (bool) -- Whether to format the output SQL.

Return type:

str

Returns:

SQL string in target dialect.

Connection Context and Session Factories#

Context managers that manage pool connection and session lifecycles for driver adapters.

class sqlspec.driver.SyncPoolConnectionContext[source]#

Bases: object

Generic sync connection context using pool.get_connection() pattern.

Subclass per adapter for type-safe provide_connection() return annotations.

__init__(config)[source]#
class sqlspec.driver.AsyncPoolConnectionContext[source]#

Bases: object

Base async connection context using pool acquire/release pattern.

Subclass per adapter and override __aenter__/__aexit__ for adapter-specific pool acquisition and release logic.

__init__(config)[source]#
class sqlspec.driver.SyncPoolSessionFactory[source]#

Bases: object

Generic sync session factory using pool.get_connection() pattern.

Subclass per adapter for type-safe acquire_connection() return annotations.

__init__(config)[source]#
class sqlspec.driver.AsyncPoolSessionFactory[source]#

Bases: object

Base async session factory using pool acquire/release pattern.

Subclass per adapter and override acquire_connection/release_connection for adapter-specific pool acquisition and release logic.

__init__(config)[source]#

Row Streaming and Execution Results#

class sqlspec.driver.SyncRowStream[source]#

Bases: Generic[RowT]

Bounded-memory iterator backed by a chunk source.

__init__(source, schema_type=None)[source]#
class sqlspec.driver.AsyncRowStream[source]#

Bases: Generic[RowT]

Async bounded-memory iterator backed by an async chunk source.

__init__(source, schema_type=None)[source]#
class sqlspec.driver.ExecutionResult[source]#

Bases: NamedTuple

Execution result containing all data needed for SQLResult building.

cursor_result: Any#

Alias for field number 0

rowcount_override: int | None#

Alias for field number 1

special_data: Any#

Alias for field number 2

selected_data: list[Any] | None#

Alias for field number 3

column_names: list[str] | None#

Alias for field number 4

data_row_count: int | None#

Alias for field number 5

statement_count: int | None#

Alias for field number 6

successful_statements: int | None#

Alias for field number 7

is_script_result: bool#

Alias for field number 8

is_select_result: bool#

Alias for field number 9

is_many_result: bool#

Alias for field number 10

row_format: Literal['dict', 'tuple', 'record']#

Alias for field number 11

last_inserted_id: int | str | None#

Alias for field number 12

column_types: dict[str, str] | None#

Alias for field number 13

static __new__(_cls, cursor_result: Any, rowcount_override: int | None, special_data: Any, selected_data: list[Any] | None, column_names: list[str] | None, data_row_count: int | None, statement_count: int | None, successful_statements: int | None, is_script_result: bool, is_select_result: bool, is_many_result: bool, row_format: Literal['dict', 'tuple', 'record'] = 'dict', last_inserted_id: int | str | None = None, column_types: dict[str, str] | None = None)#

Create new instance of ExecutionResult(cursor_result, rowcount_override, special_data, selected_data, column_names, data_row_count, statement_count, successful_statements, is_script_result, is_select_result, is_many_result, row_format, last_inserted_id, column_types)

Exception Handlers#

class sqlspec.driver.BaseSyncExceptionHandler[source]#

Bases: object

Base sync exception handler using the deferred exception pattern.

__init__()[source]#
class sqlspec.driver.BaseAsyncExceptionHandler[source]#

Bases: object

Base async exception handler using the deferred exception pattern.

__init__()[source]#

Data Dictionary#

The shared data dictionary base classes define the metadata contract used by adapter-local dictionaries. User-facing examples and the support matrix live in Data Dictionary. In short:

  • Structural metadata returns MetadataResult envelopes.

  • DDL lookups return DDLResult objects with fidelity and warning metadata.

  • Dependency ordering uses typed dependency edges rather than only table names.

  • System and performance metadata uses SystemMetadataRequest and SystemMetadataResult in a separate opt-in namespace.

class sqlspec.driver.DataDictionaryMixin[source]#

Bases: object

Mixin providing common data dictionary functionality.

Includes version caching to avoid repeated database queries when checking feature flags or optimal types.

dialect: ClassVar[str]#
get_cached_version(driver_id)[source]#

Get cached version info for a driver.

Parameters:

driver_id (int) -- The id() of the driver instance.

Return type:

tuple[bool, VersionInfo | None]

Returns:

Tuple of (was_cached, version_info). If was_cached is False, the caller should fetch the version and call cache_version().

cache_version(driver_id, version)[source]#

Cache version info for a driver.

Parameters:

  • driver_id (int) -- The id() of the driver instance.

  • version (VersionInfo | None) -- The version info to cache (can be None if detection failed).

Return type:

None

parse_version_with_pattern(pattern, version_str)[source]#

Parse version string using a specific regex pattern.

Parameters:

  • pattern (Pattern[str]) -- Compiled regex pattern for the version format

  • version_str (str) -- Raw version string from database

Return type:

VersionInfo | None

Returns:

VersionInfo instance or None if parsing fails

get_default_type_mapping()[source]#

Get default type mappings for common categories.

Return type:

dict[str, str]

Returns:

Dictionary mapping type categories to generic SQL types

sort_tables_topologically(tables, foreign_keys)[source]#

Sort tables topologically based on typed foreign key dependency edges.

Parameters:
Return type:

list[str]

Returns:

List of table names in topological order (dependencies first).

class sqlspec.driver.AsyncDataDictionaryBase[source]#

Bases: DataDictionaryDialectMixin, DataDictionaryMixin

Base class for asynchronous data dictionary implementations.

Uses Python-compatible class layouts for cross-module inheritance. Child classes define dialect as a class attribute.

dialect: ClassVar[str]#

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

__init__()[source]#
async get_metadata_capabilities(driver, domains=None)[source]#

Get data-dictionary capability profile.

Parameters:

  • driver (Any) -- Async database driver instance.

  • domains (Sequence[str] | None) -- Optional metadata domains to report.

Return type:

MetadataCapabilityProfile

Returns:

Capability profile. Base implementations report requested domains as unsupported.

async get_system_metadata_capabilities(driver, domains=None)[source]#

Get opt-in system metadata capability disclosures.

Parameters:

  • driver (Any) -- Async database driver instance.

  • domains (Sequence[str] | None) -- Optional system metadata domains to report.

Return type:

tuple[SystemMetadataCapability, ...]

Returns:

System metadata capabilities. Base implementations report requested domains as unsupported.

async get_schemas(driver)[source]#

Get schema metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get database object metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get rich table metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get constraint metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get view metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get routine metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get privilege metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get dependency metadata or an unsupported-domain result.

Return type:

MetadataResult

async get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Get object DDL or an explicit unsupported DDL result.

Return type:

DDLResult

async get_object_ddl(driver, object_name, *, schema=None, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Get one object's DDL or an explicit unsupported DDL result.

Return type:

DDLResult

async get_schema_ddl(driver, schema=None, *, include_domains=None, exclude_domains=None, include_dependencies=True, prefer_native=True, redact=True)[source]#

Get schema DDL items or an unsupported-domain result.

Return type:

MetadataResult

async get_system_metadata(driver, request=None, **kwargs)[source]#

Get opt-in system metadata or a gated/unsupported result.

Return type:

SystemMetadataResult

abstractmethod async get_version(driver)[source]#

Get database version information.

Parameters:

driver (Any) -- Async database driver instance

Return type:

VersionInfo | None

Returns:

Version information or None if detection fails

abstractmethod async get_feature_flag(driver, feature)[source]#

Check if database supports a specific feature.

Parameters:

  • driver (Any) -- Async database driver instance

  • feature (str) -- Feature name to check

Return type:

bool

Returns:

True if feature is supported, False otherwise

abstractmethod async get_optimal_type(driver, type_category)[source]#

Get optimal database type for a category.

Parameters:

  • driver (Any) -- Async database driver instance

  • type_category (str) -- Type category

Return type:

str

Returns:

Database-specific type name

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

Get list of tables in schema.

Parameters:

  • driver (Any) -- Async database driver instance

  • schema (str | None) -- Schema name (None for default)

Return type:

list[TableMetadata]

Returns:

List of table metadata dictionaries

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

Get column information for a table or schema.

Parameters:

  • driver (Any) -- Async database driver instance

  • table (str | None) -- Table name (None to fetch columns for all tables in schema)

  • schema (str | None) -- Schema name (None for default)

Return type:

list[ColumnMetadata]

Returns:

List of column metadata dictionaries

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

Get index information for a table or schema.

Parameters:

  • driver (Any) -- Async database driver instance

  • table (str | None) -- Table name (None to fetch indexes for all tables in schema)

  • schema (str | None) -- Schema name (None for default)

Return type:

list[IndexMetadata]

Returns:

List of index metadata dictionaries

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

Get foreign key metadata.

Parameters:

  • driver (Any) -- Async database driver instance

  • table (str | None) -- Optional table name filter

  • schema (str | None) -- Optional schema name filter

Return type:

list[ForeignKeyMetadata]

Returns:

List of foreign key metadata

class sqlspec.driver.SyncDataDictionaryBase[source]#

Bases: DataDictionaryDialectMixin, DataDictionaryMixin

Base class for synchronous data dictionary implementations.

Uses Python-compatible class layouts for cross-module inheritance. Child classes define dialect as a class attribute.

dialect: ClassVar[str]#

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

__init__()[source]#
get_metadata_capabilities(driver, domains=None)[source]#

Get data-dictionary capability profile.

Parameters:

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

  • domains (Sequence[str] | None) -- Optional metadata domains to report.

Return type:

MetadataCapabilityProfile

Returns:

Capability profile. Base implementations report requested domains as unsupported.

get_system_metadata_capabilities(driver, domains=None)[source]#

Get opt-in system metadata capability disclosures.

Parameters:

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

  • domains (Sequence[str] | None) -- Optional system metadata domains to report.

Return type:

tuple[SystemMetadataCapability, ...]

Returns:

System metadata capabilities. Base implementations report requested domains as unsupported.

get_schemas(driver)[source]#

Get schema metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get database object metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get rich table metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get constraint metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get view metadata or an unsupported-domain result.

Return type:

MetadataResult

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

Get routine metadata or an unsupported-domain result.

Return type:

MetadataResult

get_privileges(driver, object_name=None, schema=None)[source]#

Get privilege metadata or an unsupported-domain result.

Return type:

MetadataResult

get_dependencies(driver, object_name=None, schema=None)[source]#

Get dependency metadata or an unsupported-domain result.

Return type:

MetadataResult

get_ddl(driver, object_name, schema=None, *, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Get object DDL or an explicit unsupported DDL result.

Return type:

DDLResult

get_object_ddl(driver, object_name, *, schema=None, object_type='table', include_dependencies=True, prefer_native=True, redact=True)[source]#

Get one object's DDL or an explicit unsupported DDL result.

Return type:

DDLResult

get_schema_ddl(driver, schema=None, *, include_domains=None, exclude_domains=None, include_dependencies=True, prefer_native=True, redact=True)[source]#

Get schema DDL items or an unsupported-domain result.

Return type:

MetadataResult

get_system_metadata(driver, request=None, **kwargs)[source]#

Get opt-in system metadata or a gated/unsupported result.

Return type:

SystemMetadataResult

abstractmethod get_version(driver)[source]#

Get database version information.

Parameters:

driver (Any) -- Sync database driver instance

Return type:

VersionInfo | None

Returns:

Version information or None if detection fails

abstractmethod get_feature_flag(driver, feature)[source]#

Check if database supports a specific feature.

Parameters:

  • driver (Any) -- Sync database driver instance

  • feature (str) -- Feature name to check

Return type:

bool

Returns:

True if feature is supported, False otherwise

abstractmethod get_optimal_type(driver, type_category)[source]#

Get optimal database type for a category.

Parameters:

  • driver (Any) -- Sync database driver instance

  • type_category (str) -- Type category

Return type:

str

Returns:

Database-specific type name

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

Get list of tables in schema.

Parameters:

  • driver (Any) -- Sync database driver instance

  • schema (str | None) -- Schema name (None for default)

Return type:

list[TableMetadata]

Returns:

List of table metadata dictionaries

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

Get column information for a table or schema.

Parameters:

  • driver (Any) -- Sync database driver instance

  • table (str | None) -- Table name (None to fetch columns for all tables in schema)

  • schema (str | None) -- Schema name (None for default)

Return type:

list[ColumnMetadata]

Returns:

List of column metadata dictionaries

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

Get index information for a table or schema.

Parameters:

  • driver (Any) -- Sync database driver instance

  • table (str | None) -- Table name (None to fetch indexes for all tables in schema)

  • schema (str | None) -- Schema name (None for default)

Return type:

list[IndexMetadata]

Returns:

List of index metadata dictionaries

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

Get foreign key metadata.

Parameters:

  • driver (Any) -- Sync database driver instance

  • table (str | None) -- Optional table name filter

  • schema (str | None) -- Optional schema name filter

Return type:

list[ForeignKeyMetadata]

Returns:

List of foreign key metadata

Adapter Data Dictionary Classes#

Adapter data dictionaries remain public at their adapter-local import paths, and drivers continue to expose them through driver.data_dictionary. Shared helper modules under sqlspec.data_dictionary.dialects are internal implementation details used to keep repeated dialect rules consistent. Performance builds compile the shared helper modules, while adapter-local data-dictionary classes stay in their driver packages because they still own real adapter overrides.

When changing data-dictionary behavior, review these shared-dialect groups together:

  • PostgreSQL: sqlspec.adapters.adbc.data_dictionary.AdbcDataDictionary when the driver dialect is Postgres, plus sqlspec.adapters.asyncpg.data_dictionary.AsyncpgDataDictionary, sqlspec.adapters.psqlpy.data_dictionary.PsqlpyDataDictionary, sqlspec.adapters.psycopg.data_dictionary.PsycopgSyncDataDictionary, and sqlspec.adapters.psycopg.data_dictionary.PsycopgAsyncDataDictionary.

  • SQLite: sqlspec.adapters.sqlite.data_dictionary.SqliteDataDictionary, sqlspec.adapters.aiosqlite.data_dictionary.AiosqliteDataDictionary, and sqlspec.adapters.adbc.data_dictionary.AdbcDataDictionary when the driver dialect is SQLite.

  • MySQL and MariaDB: sqlspec.adapters.mysqlconnector.data_dictionary.MysqlConnectorSyncDataDictionary, sqlspec.adapters.mysqlconnector.data_dictionary.MysqlConnectorAsyncDataDictionary, sqlspec.adapters.pymysql.data_dictionary.PyMysqlDataDictionary, sqlspec.adapters.aiomysql.data_dictionary.AiomysqlDataDictionary, sqlspec.adapters.asyncmy.data_dictionary.AsyncmyDataDictionary, and sqlspec.adapters.adbc.data_dictionary.AdbcDataDictionary when the driver dialect is MySQL or MariaDB.

  • CockroachDB: sqlspec.adapters.cockroach_asyncpg.data_dictionary.CockroachAsyncpgDataDictionary, sqlspec.adapters.cockroach_psycopg.data_dictionary.CockroachPsycopgSyncDataDictionary, sqlspec.adapters.cockroach_psycopg.data_dictionary.CockroachPsycopgAsyncDataDictionary, and sqlspec.adapters.adbc.data_dictionary.AdbcDataDictionary when the driver dialect is CockroachDB.

  • Oracle: sqlspec.adapters.oracledb.data_dictionary.OracledbSyncDataDictionary and sqlspec.adapters.oracledb.data_dictionary.OracledbAsyncDataDictionary. Oracle ADK and event stores that construct these dictionaries directly should be reviewed with the same changes.

  • BigQuery: sqlspec.adapters.bigquery.data_dictionary.BigQueryDataDictionary and sqlspec.adapters.adbc.data_dictionary.AdbcDataDictionary when the driver dialect is BigQuery.

  • DuckDB and Spanner currently have native-only adapter dictionaries: sqlspec.adapters.duckdb.data_dictionary.DuckDBDataDictionary and sqlspec.adapters.spanner.data_dictionary.SpannerDataDictionary.

Feature Flag Types#

class sqlspec.data_dictionary.FeatureFlags[source]

Bases: TypedDict

Typed feature flags for data dictionary dialects.

supports_arrays: bool
supports_clustering: bool
supports_crdb_internal_metadata: bool
supports_cte: bool
supports_events: bool
supports_generated_columns: bool
supports_generators: bool
supports_for_update: bool
supports_geography: bool
supports_in_memory: bool
supports_index_clustering: bool
supports_invisible_columns: bool
supports_invisible_indexes: bool
supports_interleaved_tables: bool
supports_json: bool
supports_maps: bool
supports_on_conflict: bool
supports_partitioning: bool
supports_prepared_statements: bool
supports_resource_groups: bool
supports_roles: bool
supports_sequences: bool
supports_system_versioned_tables: bool
supports_returning: bool
supports_schemas: bool
supports_skip_locked: bool
supports_structs: bool
supports_transactions: bool
supports_update_from: bool
supports_upsert: bool
supports_uuid: bool
supports_window_functions: bool
class sqlspec.data_dictionary.FeatureVersions[source]

Bases: TypedDict

Typed feature version requirements for data dictionary dialects.

supports_cte: VersionInfo
supports_check_constraints: VersionInfo
supports_events: VersionInfo
supports_generated_columns: VersionInfo
supports_histograms: VersionInfo
supports_invisible_columns: VersionInfo
supports_invisible_indexes: VersionInfo
supports_json: VersionInfo
supports_jsonb: VersionInfo
supports_partitioning: VersionInfo
supports_resource_groups: VersionInfo
supports_returning: VersionInfo
supports_roles: VersionInfo
supports_sequences: VersionInfo
supports_skip_locked: VersionInfo
supports_system_versioned_tables: VersionInfo
supports_upsert: VersionInfo
supports_window_functions: VersionInfo