Quickstart

This quickstart guide will get you running your first SQLSpec query in under 5 minutes. We’ll use SQLite since it requires no additional setup, but the same patterns apply to all supported databases.

Your First Query

Let’s start with the simplest possible example - executing a query and getting results:

first sqlspec query
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))

with db_manager.provide_session(db) as session:
    result = session.execute("SELECT 'Hello, SQLSpec!' as message")
    print(result.get_first())

What’s happening here?

  1. Create SQLSpec instance: This is your central registry for database configurations.

  2. Configure a database: Each database gets a config object. Here we’re using in-memory SQLite.

  3. Get a session: The context manager provides a database session and handles cleanup.

  4. Execute SQL: Write your SQL directly - no magic, no abstraction.

Working with Real Data

Let’s create a table, insert some data, and query it:

working with real data
from sqlspec import SQLSpec
from sqlspec.adapters.sqlite import SqliteConfig

db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))

with db_manager.provide_session(db) as session:
    _ = session.execute(
        """
        CREATE TABLE users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL
        )
        """
    )

    _ = session.execute("INSERT INTO users (name, email) VALUES (?, ?)", "Alice", "[email protected]")

    _ = session.execute_many(
        "INSERT INTO users (name, email) VALUES (?, ?)",
        [("Bob", "[email protected]"), ("Charlie", "[email protected]")],
    )

    users = session.select("SELECT * FROM users")
    print(f"All users: {users}")

    alice = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Alice")
    print(f"Alice: {alice}")

    count = session.select_value("SELECT COUNT(*) FROM users")
    print(f"Total users: {count}")

Session Methods Cheat Sheet

SQLSpec provides several convenient query methods:

  • execute(sql, *params): Execute any SQL, returns full result

  • select(sql, *params): Execute SELECT, returns list of rows

  • select_one(sql, *params): Get single row, raises error if not found

  • select_one_or_none(sql, *params): Get single row or None

  • select_value(sql, *params): Get single scalar value

  • execute_many(sql, params_list): Execute with multiple parameter sets

  • execute_script(sql): Execute multiple SQL statements

Type-Safe Results

The real power of SQLSpec comes from type-safe result mapping. Define your data models and SQLSpec automatically maps query results to them:

type-safe results
from pydantic import BaseModel

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

class User(BaseModel):
    id: int
    name: str
    email: str

db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))

with db_manager.provide_session(db) as session:
    _ = session.execute(
        """
        CREATE TABLE users (id INTEGER, name TEXT, email TEXT)
        """
    )
    _ = session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "[email protected]")

    user = session.select_one("SELECT * FROM users WHERE id = ?", 1, schema_type=User)
    print(f"User: {user.name} ({user.email})")

    all_users = session.select("SELECT * FROM users", schema_type=User)
    for typed_user in all_users:
        print(f"User: {typed_user.name}")

Note

SQLSpec supports multiple type libraries: Pydantic, msgspec, attrs, and standard library dataclasses. Choose the one that fits your project!

Async Support

SQLSpec supports async/await for non-blocking database operations. Here’s the same example with async:

async support
from pydantic import BaseModel

from sqlspec import SQLSpec
from sqlspec.adapters.aiosqlite import AiosqliteConfig

class User(BaseModel):
    id: int
    name: str
    email: str

db_manager = SQLSpec()
db = db_manager.add_config(AiosqliteConfig(pool_config={"database": ":memory:"}))

async with db_manager.provide_session(db) as session:
    await session.execute(
        """
        CREATE TABLE users (id INTEGER, name TEXT, email TEXT)
        """
    )
    await session.execute("INSERT INTO users VALUES (?, ?, ?)", 1, "Alice", "[email protected]")
    user = await session.select_one("SELECT * FROM users WHERE id = ?", 1, schema_type=User)
    print(f"User: {user.name}")

The API is identical - just add await and use async config/drivers!

Switching Databases

One of SQLSpec’s strengths is the consistent API across databases. Here’s the same code using PostgreSQL:

switching databases
import os
from typing import Any

from pydantic import BaseModel

from sqlspec import SQLSpec
from sqlspec.adapters.asyncpg import AsyncpgConfig

class User(BaseModel):
    id: int
    name: str
    email: str

def pool_config() -> "dict[str, Any]":
    return {
        "host": os.getenv("SQLSPEC_QUICKSTART_PG_HOST", "localhost"),
        "port": int(os.getenv("SQLSPEC_QUICKSTART_PG_PORT", "5432")),
        "user": os.getenv("SQLSPEC_QUICKSTART_PG_USER", "postgres"),
        "password": os.getenv("SQLSPEC_QUICKSTART_PG_PASSWORD", "postgres"),
        "database": os.getenv("SQLSPEC_QUICKSTART_PG_DATABASE", "mydb"),
    }

async def seed_users(session: Any) -> None:
    await session.execute(
        """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            email TEXT NOT NULL
        )
        """
    )
    await session.execute("TRUNCATE TABLE users")
    await session.execute(
        "INSERT INTO users (id, name, email) VALUES ($1, $2, $3)", 1, "Alice", "[email protected]"
    )

db_manager = SQLSpec()
db = db_manager.add_config(AsyncpgConfig(pool_config=pool_config()))

async with db_manager.provide_session(db) as session:
    await seed_users(session)
    user = await session.select_one("SELECT * FROM users WHERE id = $1", 1, schema_type=User)
    print(f"User: {user.name}")

Note

For tests we surface the PostgreSQL connection info through the SQLSPEC_QUICKSTART_PG_* environment variables (host, port, user, password, database). When running the snippet outside pytest, export those variables or inline your DSN in _pool_config() so the example keeps working.

Tip

Each database has its own parameter style (? for SQLite, $1 for PostgreSQL, %s for MySQL, etc.). SQLSpec handles this automatically - you just need to use the correct style for your database.

PEP249 is worth a read on that

Multiple Databases

Need to work with multiple databases? Register multiple configs:

multiple databases
from sqlspec import SQLSpec
from sqlspec.adapters.duckdb import DuckDBConfig
from sqlspec.adapters.sqlite import SqliteConfig

db_manager = SQLSpec()
sqlite_db = db_manager.add_config(SqliteConfig(pool_config={"database": "app.db"}))
duckdb_db = db_manager.add_config(DuckDBConfig(pool_config={"database": "analytics.duckdb"}))

with db_manager.provide_session(sqlite_db) as sqlite_session:
    users = sqlite_session.select("SELECT 1")

with db_manager.provide_session(duckdb_db) as duckdb_session:
    analytics = duckdb_session.select("SELECT 1")

Transaction Support

Sessions expose begin(), commit(), and rollback() so you can control transaction boundaries explicitly and keep examples deterministic:

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

db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))

with db_manager.provide_session(db) as session:
    session.begin()
    session.execute(
        """
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL
        )
        """
    )
    session.execute(
        """
        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY,
            user_name TEXT NOT NULL
        )
        """
    )
    session.execute("DELETE FROM users")
    session.execute("DELETE FROM orders")
    session.execute("INSERT INTO users (name) VALUES (?)", "Alice")
    session.execute("INSERT INTO orders (user_name) VALUES (?)", "Alice")
    session.commit()

with db_manager.provide_session(db) as session:
    session.begin()
    session.execute("INSERT INTO users (name) VALUES (?)", "Bob")
    session.rollback()

with db_manager.provide_session(db) as session:
    alice = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Alice")
    bob = session.select_one_or_none("SELECT * FROM users WHERE name = ?", "Bob")

The snippet seeds data in a temporary SQLite database, intentionally triggers a failure, and uses contextlib.suppress so the docs stay readable while the companion test verifies that rollback() removes the failed insert.

Note

Transaction behavior can be configured per session or globally. See the Drivers and Querying guide for details on transaction modes.

Query Builder (Experimental)

For those who prefer programmatic query construction, SQLSpec includes an experimental query builder:

query builder
from sqlspec import SQLSpec, sql
from sqlspec.adapters.sqlite import SqliteConfig

query = sql.select("id", "name", "email").from_("users").where("age > ?").order_by("name")

db_manager = SQLSpec()
db = db_manager.add_config(SqliteConfig(pool_config={"database": ":memory:"}))

with db_manager.provide_session(db) as session:
    session.execute(
        """
        CREATE TABLE users (id INTEGER, name TEXT, email TEXT, age INTEGER)
        """
    )
    session.execute("INSERT INTO users VALUES (?, ?, ?, ?)", 1, "Alice", "[email protected]", 30)
    results = session.select(query, 25)
    print(results)

Warning

The query builder API is experimental and will change significantly in future releases. Use raw SQL for production code.

Next Steps

You’ve now seen the basics of SQLSpec! Here’s where to go next:

Usage Guides

Examples

API Reference

Tip

SQLSpec is designed to be simple but powerful. Start with raw SQL and add features like type-safe mapping and the query builder only when you need them.