Drivers and Querying

SQLSpec provides a unified driver interface across supported databases. This page covers the common execution patterns and points to adapter-specific configuration.

Supported Drivers (High Level)

  • PostgreSQL: asyncpg, psycopg (sync/async), psqlpy, ADBC

  • SQLite: sqlite3, aiosqlite, ADBC

  • MySQL: asyncmy, mysql-connector, pymysql

  • Analytics / Cloud: DuckDB, BigQuery, Spanner, Oracle, ADBC

Core Execution Pattern

sqlite session
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

db_path = tmp_path / "driver.sqlite"
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 health (id integer primary key)")
    result = session.execute("select count(*) as total from health")
    print(result.one())

Transactions

manual transaction
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

db_path = tmp_path / "transactions.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 ledger (id integer primary key, note text)")
    session.begin()
    session.execute("insert into ledger (note) values ('committed')")
    session.commit()

    session.begin()
    session.execute("insert into ledger (note) values ('rolled back')")
    session.rollback()

    result = session.execute("select note from ledger order by id")
    print(result.all())

Parameter Binding

positional and named parameters
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

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

with spec.provide_session(config) as session:
    session.execute("select :status as status, :status as status_copy", {"status": "active"})
    result = session.execute("select ? as value", (42,))
    print(result.one())

Statement Stacks

Statement stacks bundle multiple SQL statements plus parameter sets. Drivers that support native pipelines or batch execution can send the stack in a single round trip, while others execute each statement sequentially.

statement stack
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig
from sqlspec.core.stack import StatementStack

spec = SQLSpec()
config = spec.add_config(SqliteConfig(connection_config={"database": ":memory:"}))

stack = (
    StatementStack()
    .push_execute("create table teams (id integer primary key, name text)")
    .push_execute_many("insert into teams (name) values (:name)", [{"name": "Litestar"}, {"name": "SQLSpec"}])
    .push_execute("select id, name from teams order by id")
)

with spec.provide_session(config) as session:
    results = session.execute_stack(stack)
    rows = results[-1].result.all()

Driver Configuration Examples

asyncpg config
from sqlspec.adapters.asyncpg import AsyncpgConfig

config = AsyncpgConfig(
    connection_config={"dsn": "postgresql://user:pass@localhost:5432/app"},
    pool_config={"min_size": 1, "max_size": 5},
)
cockroach + psycopg
from sqlspec.adapters.cockroach_psycopg import CockroachPsycopgSyncConfig

config = CockroachPsycopgSyncConfig(connection_config={"dsn": "postgresql://user:pass@localhost:26257/defaultdb"})
mysql connector config
from sqlspec.adapters.mysqlconnector import MysqlConnectorSyncConfig

config = MysqlConnectorSyncConfig(
    connection_config={"host": "localhost", "user": "app", "password": "secret", "database": "app_db"}
)